使用 JEP 将数据帧从 scala 传递到 python

Question

这是我正在尝试做的：

1. 我将数据读入 scala
2. 提取几列
3. 使用 JEP 将创建的数据帧传递给 Python 脚本
4. Python 脚本将数据帧转换为熊猫执行一些操作并将其返回

但是我不确定如何将数据帧传递给 python 脚本。 这是python脚本(这只是示例脚本，而不是实际脚本)：

import findspark
findspark.init()
import pandas as pd
#from pyspark.sql import types.*
from pyspark.sql import DataFrame as dataframe
def tes(df: dataframe):
df = df.toPandas()
df['concatenate'] = df['country'] + df['datasourceProvidedCountry']
return dataframe(df)

并且它不断失败并出现以下错误：

jep.JepException: <class 'ImportError'>: py4j.protocol
at /usr/local/lib64/python3.6/site-packages/jep/java_import_hook.__getattr__(java_import_hook.py:57)
at /home/hadoop/testpy.<module>(testpy.py:5)
at jep.Jep.run(Native Method)
at jep.Jep.runScript(Jep.java:359)
at jep.Jep.runScript(Jep.java:335)
... 49 elided
Caused by: java.lang.ClassNotFoundException: py4j.protocol
at java.net.URLClassLoader.findClass(URLClassLoader.java:382)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
... 52 more

spark-shell --conf spark.driver.extraLibraryPath=:/usr/local/lib64/python3.6/site-packages/jep:/usr/local/lib/python3.6/site-packages/py4j/ --jars /home/hadoop/jep-3.8.2.jar

谁能建议如何使用 Jep 将数据帧从 scala 传递到 pyspark(如果这是重复的，请指出我正确的线程，因为我找不到)？

Answer 1

我有同样的要求，也尝试了 Jep。不幸的是，Jep 不适用于此用例。

未找到的py4j.protocol是由Jep ClassEnquirer引起的，当python和jave都有同名的库时，Jep会考虑java库。您可以通过从 java 应用程序中从 Spark 包中排除 py4j 来解决此问题，或者创建一个自定义的 ClassEnquirer 来考虑 python py4j。

您还需要更新 Jep 构造函数，将 useSubInterpreter 值设置为 false 并重建它。

public Jep(JepConfig config) throws JepException {
this(config, false);
}

现在应该解决了错误。但是，传递给python函数的对象是包含java引用的PyObject，它不是pyspark数据帧对象，因此它没有toPandas()函数。

另一种方法可能是使用 gRPC 或 Apache 节俭，您可以查看文档以获取更多详细信息。

Answer 2

可以使用Apache Arrow将数据从Apache Spark正确的(JVM)传递到Python代码 - 因为2.3 Spark使用可以从JVM和CPython使用的Arrow格式。

请参阅 https://fossies.org/diffs/spark/2.3.3_vs_2.4.0/sql/core/src/test/scala/org/apache/spark/sql/execution/arrow/ArrowConvertersSuite.scala-diff.html 以获取灵感。

我使用相同的进程(无套接字)在JVM和CPython代码之间传递数据，使用jep(Java Embedded Python)DirectNDArray("off-heap"，"zero copy")。

请让我知道这看起来是否足够好，并将改善这个答案。