在pyspark中调用scala代码进行XSLT转换

Question

这可能是一个长镜头,但认为它可以不会受到伤害.我试图在pyspark中使用Elsevier的开源spark-xml-utils软件包来使用XSLT转换一些XML记录.

我通过一些探索性代码获得转换工作取得了一些成功:

# open XSLT processor from spark's jvm context
with open('/tmp/foo.xsl', 'r') as f:
    proc = sc._jvm.com.elsevier.spark_xml_utils.xslt.XSLTProcessor.getInstance(f.read())


# transform XML record with 'proc' 
with open('/tmp/bar.xml','r') as f:
    transformed = proc.transform(f.read())

但是,在更现实的情况下,我无法proc.transform进入lambda地图功能,得到的错误类似于:

"调用o55时发生错误.getstate.跟踪:py4j.Py4JException:方法getstate([])在py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:318)py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine)中不存在.java:326)py4j.Gateway.invoke(Gateway.java:272)py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)at py4j.commands.CallCommand.execute(CallCommand.java:79)py4j .gatewayConnection.run(GatewayConnection.java:214)at java.lang.Thread.run(Thread.java:748)"

当我得到一个小例子来处理单个记录时,那是在pyspark shell中运行的,我假设它正在使用火花驱动器.但是在上面提到的map函数中,这是通过Livy和YARN在Spark中引入的工作者.这个问题/答案表明,在这种情况下,我可能无法使用jvm中的函数.

现在,该spark-xml-utils库提供了一些scala示例,正是我想要做的事情:

import com.elsevier.spark_xml_utils.xslt.XSLTProcessor

val xmlKeyPair = sc.sequenceFile[String, String]("s3n://spark-xml-utils/xml/part*")

val stylesheet = sc.textFile("s3n://spark-xml-utils/stylesheets/srctitle.xsl").collect.head

val srctitles = xmlKeyPair.mapPartitions(recsIter => { 
    val proc = XSLTProcessor.getInstance(stylesheet)
    recsIter.map(rec => proc.transform(rec._2))
})

我想知道,我怎样才能将其转换为pyspark代码,以便我可以在RDD上运行它？理想情况下,在具有以下输入和输出格式的RDD上:

id | document | other | columns
-----------------------------------------------------
sprog | <xml here...> | more | data
baz   | <xml here...> | more | data

可能会成为

id | document | other | columns
-----------------------------------------------------
sprog | <*transformed* xml here...> | more | data
baz   | <*transformed* xml here...> | more | data

任何帮助或建议将非常感激.

更新8/28/2018:也尝试过mapPartitions,没有骰子.同样的错误__getstate__()