如何将 numpy 数组元素转换为 spark RDD 列值答案

【问题标题】：How to convert numpy array elements to spark RDD column values如何将 numpy 数组元素转换为 spark RDD 列值
【发布时间】：2018-06-11 19:01:15
【问题描述】：

我正准备使用 spark 数据框（不是 pandas）的内置 CSV 打印工具。我已经建立了一个 IndexedRowMatrix。因此，其中有稀疏的数组列。通过 x.vector.toArray() 映射调用（如下）执行到密集数组的转换。我似乎无法弄清楚，如何将密集的 numpy 数组元素放入 Spark DataFrame 的各个列中。（请不要介绍熊猫。）如何将此 RDD 放入 7 列数据框，由一个字符串列和六个整数列组成？到目前为止我的代码：

X = CoordinateMatrix(sc.parallelize(entries)) 
Xirm = X.toIndexedRowMatrix()
colnames = "username," + ','.join(str(cell) for cell in itemids.keys())  # Make CSV header line
# Might need this for speed: Arrow:  https://bryancutler.github.io/createDataFrame/  See above conf=...
XX = Xirm.rows.map(lambda x: (lu[x.index], x.vector.toArray())) # ?
print(XX.take(2))
df = XX.toDF() #TypeError: Can not infer schema for type: <class 'numpy.ndarray'>
#df.write.csv(header=colnames, path=out_filename)

这里是take(2)查看数据示例：

[('kygiacomo', array([ 0.,  1.,  0.,  0.,  0.,  0.])), ('namohysip', array([ 1.,  0.,  0.,  0.,  0.,  0.]))]

看到问题是 RDD 元组有 2 列，但我在 DataFrame 中需要 7 列。列数是动态确定的，我在 colnames 变量中有列名，但我也不知道如何输入它。同样，目标是通过使用 DAtaFrame 的 spark 内置 CSV 写入功能，输出“等效”的 CSV 文件（许多部分文件都可以）。（Spark 2.3.0 是常驻的。）理想情况下，实数将转换为整数，并且任何数据值都没有引号。但是 2 到 7 列的转换是目前真正棘手的问题。感谢您的提示。

【问题讨论】：

标签： python numpy apache-spark

【解决方案1】：

简单的 Python 类型转换和解包应该可以解决问题：

Xirm.rows.map(lambda x: (lu[x.index], *x.vector.toArray().tolist()))

一样

Xirm.rows.map(lambda x: [lu[x.index]] + x.vector.toArray().tolist())

【讨论】：