在 pyspark -pandas_udf 中按组应用函数（没有名为 pyarrow 的模块）

Question

我正在尝试对 pyspark 中的每个数据集组应用一个函数。我遇到的第一个错误是

Py4JError: An error occurred while calling o62.__getnewargs__. Trace:
py4j.Py4JException: Method __getnewargs__([]) does not exist

为了解决上述问题，我删除了 spark 函数（我有 spark.range()）。现在错误已解决，但我现在得到以下信息：

File "/opt/cloudera/parcels/SPARK2-2.3.0.cloudera2-1.cdh5.13.3.p0.316101/lib/spark2/python/pyspark/serializers.py", line 276, in load_stream
    import pyarrow as pa
ImportError: No module named pyarrow

但是当我自己尝试时，它起作用了，即。

df = pd.DataFrame({"a": [1, 2, 3]})
pa.Table.from_pandas(df)
pyarrow.Table
a: int64
__index_level_0__: int64
metadata
--------
{'pandas': '{"pandas_version": "0.23.0", "index_columns": ["__index_level_0__"], "columns": [{"metadata": null, "field_name": "a", "name": "a", "numpy_type": "int64", "pandas_type": "int64"}, {"metadata": null, "field_name": "__index_level_0__", "name": null, "numpy_type": "int64", "pandas_type": "int64"}], "column_indexes": [{"metadata": null, "field_name": null, "name": null, "numpy_type": "object", "pandas_type": "bytes"}]}'}

失败的例子 - 摘自here

import pyspark.sql.functions as F
import pandas as pd

cols = ['id', 'val']
vals = [('A', 5), ('A', 3), ('A', 7), ('B', 12), ('B', 15), ('C', 3)]
d1 = sqlContext.createDataFrame(vals, cols)

>>> @F.pandas_udf(d1.schema, F.PandasUDFType.GROUPED_MAP)
... def subtract_mean(pdf):
...     return pdf.assign(v=pdf.v - pdf.v.mean())
...
>>> try1 = d1.groupby('id').apply(subtract_mean)
>>> try1.show()

如果我尝试转换为 pandas 数据帧，它不起作用，因为它没有 schema 属性。

我在这里错过了什么？

Answer 1

pyarrow 必须出现在每个工作节点的路径上。

如果您运行将此代码作为单个节点，请确保PYSPARK_PYTHON（以及可选的PYTHONPATH）与您用来测试[=10]的解释器相同=]代码。
如果您使用集群，请确保 pyarrow 安装在每个节点上，此外还有上面提到的要点。

此外，请确保安装的 pyarrow 版本大于或等于支持的最低版本 (0.8 today) - 虽然这会导致不同的异常。

Answer 2

添加到已接受的答案中：

一个问题可以是用户委派：ie

python 模块安装 userx
spark-submit 由 userx
但是 spark 是技术用户在工人身上执行逻辑。
（在幕后，在你不注意的情况下）。

可能是 spark 没有看到模块。

解决这个问题：
su 为 spark，并通过 pipinstall 安装缺少的模块（对所有工人）

帮我修好了。

编辑：
根据您的设置 pyspark shell 可能仍然可以看到您的模块。

在 pyspark -pandas_udf 中按组应用函数（没有名为 pyarrow 的模块）

Apply function per group in pyspark -pandas_udf (No module named pyarrow)

python

apache-spark

pyspark

pyarrow