从 numpy 矩阵创建 Spark 数据框

Question

这是我第一次使用 PySpark (Spark 2)，我正在尝试为 Logit 模型创建一个玩具数据框。我运行成功地 tutorial 并想将我自己的数据传递给它。

我试过这个：

%pyspark
import numpy as np
from pyspark.ml.linalg import Vectors, VectorUDT
from pyspark.mllib.regression import LabeledPoint

df = np.concatenate([np.random.randint(0,2, size=(1000)), np.random.randn(1000), 3*np.random.randn(1000)+2, 6*np.random.randn(1000)-2]).reshape(1000,-1)
df = map(lambda x: LabeledPoint(x[0], Vectors.dense(x[1:])), df)

mydf = spark.createDataFrame(df,["label", "features"])

但我无法摆脱：

TypeError: Cannot convert type <class 'pyspark.ml.linalg.DenseVector'> into Vector

我正在使用向量的 ML 库并且输入是双精度数组，请问有什么问题？按照documentation应该没问题。

非常感谢。

Answer 1

您混合使用了 ML 和 MLlib 的功能，它们不一定兼容。使用 spark-ml:

时不需要 LabeledPoint

sc.version
# u'2.1.1'

import numpy as np
from pyspark.ml.linalg import Vectors

df = np.concatenate([np.random.randint(0,2, size=(1000)), np.random.randn(1000), 3*np.random.randn(1000)+2, 6*np.random.randn(1000)-2]).reshape(1000,-1)
dff = map(lambda x: (int(x[0]), Vectors.dense(x[1:])), df)

mydf = spark.createDataFrame(dff,schema=["label", "features"])

mydf.show(5)
# +-----+-------------+ 
# |label|     features| 
# +-----+-------------+ 
# |    1|[0.0,0.0,0.0]| 
# |    0|[0.0,1.0,1.0]| 
# |    0|[0.0,1.0,0.0]| 
# |    1|[0.0,0.0,1.0]| 
# |    0|[0.0,1.0,0.0]|
# +-----+-------------+

PS: 从Spark 2.0开始，spark.mllib包中基于RDD的API进入维护模式。 Spark 的主要机器学习 API 现在是 spark.ml 包中基于 DataFrame 的 API。 [ref.]

Answer 2

问题很容易解决。您同时使用 ml 和 mllib API。坚持一个。否则你会得到这个错误。

这是mllibAPI的解决方案：

import numpy as np
from pyspark.mllib.linalg import Vectors, VectorUDT
from pyspark.mllib.regression import LabeledPoint

df = np.concatenate([np.random.randint(0,2, size=(1000)), np.random.randn(1000), 3*np.random.randn(1000)+2, 6*np.random.randn(1000)-2]).reshape(1000,-1)
df = map(lambda x: LabeledPoint(x[0], Vectors.dense(x[1:])), df)

mydf = spark.createDataFrame(df,["label", "features"])

对于 ml API，您实际上不再需要 LabeledPoint。这是一个example。我建议使用 ml API 因为 mllib API 很快就会被弃用。

Answer 3

从 Numpy 到 Pandas 再到 Spark：

data = np.random.rand(4,4)
df = pd.DataFrame(data, columns=list('abcd'))
spark.createDataFrame(df).show()

输出:

+-------------------+-------------------+------------------+-------------------+
|                  a|                  b|                 c|                  d|
+-------------------+-------------------+------------------+-------------------+
| 0.8026427193838694|0.16867056812634307|0.2284873209015007|0.17141853164400833|
| 0.2559088794287595| 0.3896957084615589|0.3806810025185623| 0.9362280141470332|
|0.41313827425060257| 0.8087580640179158|0.5547653674054028| 0.5386190454838264|
| 0.2948395900484454| 0.4085807623354264|0.6814694724946697|0.32031773805256325|
+-------------------+-------------------+------------------+-------------------+

从 numpy 矩阵创建 Spark 数据框

Creating Spark dataframe from numpy matrix

numpy

apache-spark

apache-spark-sql

pyspark

apache-spark-mllib