Py4J序列化：AttributeError:“dict”对象没有属性“\u get_object_id”

Question

使用Py4J，我无法将Python字典对象解析到底层JVM实例中。在

我写了一个PySpark代码，其中我在RDD上运行一个UDF/lambda函数。我的目标是在需要连接到底层JVM的RDD的每一行上运行一段Python代码，以便从我在spark提交期间使用--jars函数部署的自定义jar（源代码在Scala中）中获取Java类实例。我试过下面的例子，但是得到了上面提到的错误。这段代码在Spark 2.3集群中的每个executor核心上运行，该集群使用Cloudera发行版。出于安全原因，稍微隐藏了代码。在

from py4j.java_gateway import JavaGateway, launch_gateway
gateway = JavaGateway().launch_gateway(classpath='custom-code.jar:scala-library-2.11.8.jar:spark-catalyst_2.11-2.3.0.cloudera3.jar:scala-reflect-2.11.8.jar', jarpath='<path-to-py4j-jar>/py4j-0.10.7.jar')
jvm = gateway.jvm
input_dict = {0: 123}
temp = jvm.com.my.code.PyUtil.fullMap(input_dict)

  File "/opt/cloudera/parcels/SPARK2-2.3.0.cloudera3-1.cdh5.13.3.p0.458809/lib/spark2/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py", line 1248, in __call__
    args_command, temp_args = self._build_args(*args)
  File "/opt/cloudera/parcels/SPARK2-2.3.0.cloudera3-1.cdh5.13.3.p0.458809/lib/spark2/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py", line 1218, in _build_args
    [get_command_part(arg, self.pool) for arg in new_args])
  File "/opt/cloudera/parcels/SPARK2-2.3.0.cloudera3-1.cdh5.13.3.p0.458809/lib/spark2/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py", line 1218, in <listcomp>
    [get_command_part(arg, self.pool) for arg in new_args])
  File "/opt/cloudera/parcels/SPARK2-2.3.0.cloudera3-1.cdh5.13.3.p0.458809/lib/spark2/python/lib/py4j-0.10.7-src.zip/py4j/protocol.py", line 298, in get_command_part
    command_part = REFERENCE_TYPE + parameter._get_object_id()
AttributeError: 'dict' object has no attribute '_get_object_id'

奇怪的是，当我运行完全相同的代码，但是调用我的自定义jar的另一个函数，它接受一个字符串而不是dict，它运行得非常好，我得到了一个py4j.java_网关.JavaObject回来，这就是我想要的。在

看一下py4j的发行说明，我知道从0.5版本开始就支持字典，所以我不确定我到底做错了什么？我怀疑网关发射有问题。在

任何方向性的帮助都将不胜感激。在

Answer 1

找到了答案，如下所示：

https://www.py4j.org/advanced_topics.html#converting-python-collections-to-java-collections

我只需要自己手动转换。在

from py4j.java_collections import MapConverter
input_dict = {0: 123}
mc_run_map_dict = MapConverter().convert(input_dict, gateway_handle._gateway_client)
temp = jvm.com.my.code.PyUtil.fullMap(mc_run_map_dict)