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

Question

这是我正在尝试做的事情：

我把数据读入scala
提取几列
使用 JEP 将创建的数据帧传递给 Python 脚本
Python 脚本将数据帧转换为 pandas 执行一些操作并 returns 返回

但是我不确定如何将数据帧传递给 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 thrift，您可以查看文档了解更多详情。

Answer 2

可以使用 Apache Arrow 将数据从 Apache Spark 适当的 (JVM) 传递到 Python 代码 - 因为 2.3 Spark 使用 Arrow 格式，可以在 JVM 和 C 中使用Python.

请参阅 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 以获取灵感。

我在同一进程（无套接字）中使用 jep（Java 嵌入式 Python）DirectNDArray 在 JVM 和 CPython 代码之间传递数据（ "off-heap", "zero copy").

如果这看起来足够好并且会改进这个答案，请告诉我。

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

Pass dataframe from scala to python using JEP

python

scala

apache-spark

jep