2

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

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

但是我不确定如何将数据框传递给 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 （如果这是重复的，请指向正确的线程，因为我找不到）？

4

2 回答 2

2

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

py4j.protocol not found是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，您可以查看文档以获取更多详细信息。

于 2019-07-18T17:36:01.680 回答
0

可以使用- 因为 2.3 Spark 使用可从 JVM 和 CPython 使用的箭头格式，将数据从Apache Spark正确的（JVM）传递到 Python 代码。Apache 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 .

jep我使用（Java Embedded Python）DirectNDArray（“堆外”，“零拷贝”）在同一进程（无套接字）中在 JVM 和 CPython 代码之间传递数据。

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

于 2020-02-12T21:01:21.680 回答