Spark - 如何在不使用 pandas 的情况下将数据帧或 rdd 转换为 spark 矩阵或 numpy 数组

Question

我有 20TB 的数据。我尝试将火花数据帧转换为火花矩阵，如下所示（Solution used found here）： 我的数据框如下所示：

+-------+---------------+--------------------+
|goodsID|customer_group|customer_phone_number|
+-------+---------------+--------------------+
|    123|          XXXXX|            XXXXXXXX|
|    432|          YYYYY|            XXXXXXXX|
+-------+---------------+--------------------+

from pyspark.mllib.linalg.distributed import IndexedRow, IndexedRowMatrix

mat = IndexedRowMatrix(mydataframe.map(lambda row: IndexedRow(*row)))
mat.numRows()
mat.numCols()

但它给了我以下错误：

Caused by: org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  File "/home/test/spark-1.6.0-bin-hadoop2.6/python/lib/pyspark.zip/pyspark/worker.py", line 111, in main
    process()
  File "/home/test/spark-1.6.0-bin-hadoop2.6/python/lib/pyspark.zip/pyspark/worker.py", line 106, in process
    serializer.dump_stream(func(split_index, iterator), outfile)
  File "/home/test/spark-1.6.0-bin-hadoop2.6/python/lib/pyspark.zip/pyspark/serializers.py", line 263, in dump_stream
    vs = list(itertools.islice(iterator, batch))
  File "/home/test/spark-1.6.0-bin-hadoop2.6/python/pyspark/rdd.py", line 1293, in takeUpToNumLeft
    yield next(iterator)
  File "<stdin>", line 1, in <lambda>
TypeError: __init__() takes exactly 3 arguments (4 given)

所以我的问题是

1. 如何在 spark 中实现这一点？
2. 另外，如何将我的数据帧转换为 numpy 数组？
3. 使用带有 spark 的 pandas 真的很糟糕吗？

Answer 1

• 输入数据的类型可能有误。向量值必须是 Double (Python float)。

• 你没有正确使用IndexedRow。它需要两个参数 - 索引和向量。如果我们假设数据的类型正确

  mat = IndexedRowMatrix(mydataframe.map(
   lambda row: IndexedRow(row[0], Vectors.dense(row[1:]))))
  

• Pandas 不好吗？对于 20TB 的数据？不是最好的选择，但there exist distributed Python libraries with similar API。