问题标签 [py4j]

我试图使用 py4j 在 pyspark 中运行 java 函数。Py4j 允许访问 JVM 中的 java 对象。我创建了另一个 JVM 实例，并且能够成功运行 java 函数。

py4j 通过 GatewayServer 实例启用此通信。

我想知道我们是否可以以某种方式访问​​ spark 的内部 JVM 来运行我的 java 函数？spark 中 py4j Gatewayserver 的入口点是什么？如何将我的函数添加到入口点？

我使用 H2O 的苏打水创建了模型。现在我想将它应用于巨大的 Spark DF（填充稀疏向量）。我使用 python 和 pyspark，pysparkling。基本上我需要在里面使用 model.predict() 函数来做地图工作。但是将数据复制到 H2O 上下文中是巨大的开销，而不是一种选择。我想我要做的是，从 h2o 模型中提取 POJO（Java 类）模型，并用它在数据框中进行映射。我的问题是：

1. 有没有更好的办法？
2. 如何为 java 类编写 pyspark 包装器，我打算只使用一种方法 .score(double[] data, double[] result)
3. 如何最大限度地重用 Spark ML 库中的包装器？

谢谢！

我需要创建一个要在 pyspark python 中使用的 UDF，它使用 java 对象进行内部计算。

如果它是一个简单的python，我会做类似的事情：

并使用以下方法调用它：

但是，我需要的函数的实现是在 java 中而不是在 python 中。我需要以某种方式包装它，以便我可以从 python 中以类似的方式调用它。

我的第一次尝试是实现 java 对象，然后将其包装在 pyspark 中的 python 中并将其转换为 UDF。因序列化错误而失败。

Java代码：

pyspark 代码：

错误：

编辑：我也尝试使 java 类可序列化，但无济于事。

我的第二次尝试是在 java 中定义 UDF，但失败了，因为我不确定如何正确包装它：

java代码：包com.test1.test2；

但我将如何使用它？我试过了：

但我得到：

我尝试使用 a.call 而不是：

但是得到了： ------------------------------------------------------------ ---------------------------- TypeError Traceback (last last call last) in () 3 a = sc._gateway.jvm.com。 test1.test2.TestClassUdf(7) 4 dfint = sqlContext.range(0,15) ----> 5 df = dfint.withColumn("a",a.call(dfint.id))

任何帮助都将不胜感激。

我一直在使用 py4j 围绕一个不太友好的 Java 库构建一个用户友好的 Python 库。在大多数情况下，这很容易，py4j 一直是一个很棒的工具。但是，我在 Python 和 Java 之间发送矩阵时遇到了障碍。

具体来说，我在 java 中有一个静态函数，它接受一个整数矩阵作为其参数：

我希望能够像这样从 Py4j 调用它：

这不起作用，这并不太令人惊讶，如果将numpy_matrix其转换为普通 python 列表的列表，它也不起作用。我曾期望解决方案是构造一个 java 数组并在函数调用之前传输数据：

现在，这段代码运行正常。但是，它需要大约两分钟才能运行。顺便说一下，我正在使用的矩阵大约是 (3 x ~300,000) 个元素。

在 Py4j 中是否有一种规范的方法可以做到这一点，而不需要大量的时间来转换矩阵？我不介意花一两秒钟，但这太慢了。如果没有为这种通信设置 Py4j，是否有适用于 Python 的 Java 互操作库？

注意：Java 库将int[][]矩阵视为不可变数组；即，它从不尝试修改它。

我在 Scala 中有一个 2d 列表，dataList我想将其转换为 Pandas DataFrame。

如果我想打印dataList，一切正常，Python 中的对象类型是

据我所知，我必须使用py4j才能访问 Python 中的集合。但是，我认为我必须使用嵌套循环来访问每个元素，dataList然后放入DataFrame. 有没有更聪明的方法可以直接将其转换为DataFramepython 列表，然后DataFrame？

我正在使用 Zeppelin 0.5.5

我正在尝试org.apache.hadoop.fs.FileUtil.unTar直接从 pyspark shell 访问。

我知道我可以访问底层虚拟机（通过 py4j）sc._jvm来执行此操作，但我很难真正连接到 hdfs（尽管我的 pyspark 会话在其他方面完全正常，并且能够在集群中针对集群内的作业运行作业） .

例如：

不幸的是，这不起作用：

爪哇：

PHP：

您可以在 PHP 代码中看到一个 php 类扩展 Java 类，set()方法是 Java 的方法，它让 PHP 类传递到 Java 端。

它可以打印“！！！！” 不是 ”....” ？

我知道 py4j 可以做到，想知道 PHP/Java Bridge 是如何做到的？

像 dataframe.show() ， sqlContext.read.json 这样的操作可以正常工作，但是大多数函数都会给出 "JavaPackage object is not callable error" 。例如：当我这样做的时候

我收到一个错误

我正在使用 apache-zeppelin 解释器并将 py4j 添加到 python 路径。

当我做

导入成功

当我尝试

在 pyspark shell 中打印

但是当我在我的解释器中尝试这个时，它会打印

一旦在 INFO 级别开始记录，我就会不断地py4j.java_gateway:Received command c on object id p0在您的日志中获取大量信息。我该如何隐藏它？

我是 Spark Streaming 的新手。在 PyCharm 中使用 PySpark 我无法通过 socketTextStream 初始化。

我收到以下错误。关于问题是什么的任何想法？

回溯（最近一次通话最后）：文件“/Users/egorkorneev/PycharmProjects/BasicStreaming/SparkConsumer.py”，第 75 行，在 dc.start_streaming() 文件“/Users/egorkorneev/PycharmProjects/BasicStreaming/SparkConsumer.py”，行40、在start_streaming items = stream.socketTextStream('localhost', '9009') 文件“/Users/egorkorneev/rnd/spark-1.6.1-bin-hadoop2.6/python/pyspark/streaming/context.py”中，第 352 行，在 socketTextStream 返回 DStream(self._jssc.socketTextStream(hostname, port, jlevel), self, File "/Library/Frameworks/Python.framework/Versions/3.5/lib/python3.5/site-packages/py4j/ java_gateway.py”，第 933 行，调用中答案，self.gateway_client，self.target_id，self.name）文件“/Library/Frameworks/Python.framework/Versions/3.5/lib/python3.5/site-packages/py4j/protocol.py”，第 316 行，在get_return_value format(target_id, ".", name, value)) py4j.protocol.Py4JError: 调用 o19.socketTextStream 时出错。跟踪：py4j.Py4JException：方法 socketTextStream([class java.lang.String, class java.lang.String, class org.apache.spark.storage.StorageLevel]) 在 py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine. java:335) at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:344) at py4j.Gateway.invoke(Gateway.java:252) at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133) at py4j。 commands.CallCommand.execute(CallCommand.java:79) 在 py4j.GatewayConnection。

火花 1.6.1