使用 Scala 转换 PySpark RDD

TL;DR - 我在 PySpark 应用程序中有看起来像字符串 DStream 的东西。我想将其作为DStream[String]到 Scala 库。不过，Py4j 不会转换字符串。

我正在开发一个 PySpark 应用程序，该应用程序使用 Spark Streaming 从 Kafka 提取数据。我的消息是字符串，我想在 Scala 代码中调用一个方法，并将其传递给DStream[String]实例。但是，我无法在 Scala 代码中接收正确的 JVM 字符串。在我看来，Python 字符串没有转换为 Java 字符串，而是被序列化了。

我的问题是：如何从 Java 字符串中取出字符串DStream object?

这是我想出的最简单的 Python 代码：

from pyspark.streaming import StreamingContext
ssc = StreamingContext(sparkContext=sc, batchDuration=int(1))

from pyspark.streaming.kafka import KafkaUtils
stream = KafkaUtils.createDirectStream(ssc, ["IN"], {"metadata.broker.list": "localhost:9092"})
values = stream.map(lambda tuple: tuple[1])

ssc._jvm.com.seigneurin.MyPythonHelper.doSomething(values._jdstream)

ssc.start()

我在 PySpark 中运行此代码，并将其传递给我的 JAR 路径：

pyspark --driver-class-path ~/path/to/my/lib-0.1.1-SNAPSHOT.jar

在 Scala 方面，我有：

package com.seigneurin

import org.apache.spark.streaming.api.java.JavaDStream

object MyPythonHelper {
  def doSomething(jdstream: JavaDStream[String]) = {
    val dstream = jdstream.dstream
    dstream.foreachRDD(rdd => {
      rdd.foreach(println)
    })
  }
}

现在，假设我将一些数据发送到 Kafka：

echo 'foo bar' | $KAFKA_HOME/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic IN

The printlnScala 代码中的语句打印如下内容：

[B@758aa4d9

我期望得到foo bar反而。

现在，如果我替换简单的printlnScala 代码中的语句如下：

rdd.foreach(v => println(v.getClass.getCanonicalName))

I get:

java.lang.ClassCastException: [B cannot be cast to java.lang.String

这表明字符串实际上是作为字节数组传递的。

如果我只是尝试将此字节数组转换为字符串（我知道我什至没有指定编码）：

      def doSomething(jdstream: JavaDStream[Array[Byte]]) = {
        val dstream = jdstream.dstream
        dstream.foreachRDD(rdd => {
          rdd.foreach(bytes => println(new String(bytes)))
        })
      }

我得到了一些东西looks像（特殊字符可能会被删除）：

�]qXfoo barqa.

这表明 Python 字符串已序列化（腌制？）。我怎样才能检索正确的 Java 字符串呢？

长话短说，没有支持的方法来做这样的事情。不要在生产中尝试这个。你已被警告过。

一般来说，除了驱动程序上的一些基本 RPC 调用之外，Spark 不会将 Py4j 用于其他任何用途，并且不会在任何其他计算机上启动 Py4j 网关。当需要时（主要是 MLlib 和 SQL 的某些部分）Spark 使用Pyrolite序列化 JVM 和 Python 之间传递的对象。

这部分 API 要么是私有的 (Scala)，要么是内部的 (Python)，因此不适合一般用途。虽然理论上您可以按批次访问它：

package dummy

import org.apache.spark.api.java.JavaRDD
import org.apache.spark.streaming.api.java.JavaDStream
import org.apache.spark.sql.DataFrame

object PythonRDDHelper {
  def go(rdd: JavaRDD[Any]) = {
    rdd.rdd.collect {
      case s: String => s
    }.take(5).foreach(println)
  }
}

完整流：

object PythonDStreamHelper {
  def go(stream: JavaDStream[Any]) = {
    stream.dstream.transform(_.collect {
      case s: String => s
    }).print
  }
}

或将各个批次暴露为DataFrames（可能是最不邪恶的选择）：

object PythonDataFrameHelper {
  def go(df: DataFrame) = {
    df.show
  }
}

并按如下方式使用这些包装器：

from pyspark.streaming import StreamingContext
from pyspark.mllib.common import _to_java_object_rdd
from pyspark.rdd import RDD

ssc = StreamingContext(spark.sparkContext, 10)
spark.catalog.listTables()

q = ssc.queueStream([sc.parallelize(["foo", "bar"]) for _ in range(10)]) 

# Reserialize RDD as Java RDD<Object> and pass 
# to Scala sink (only for output)
q.foreachRDD(lambda rdd: ssc._jvm.dummy.PythonRDDHelper.go(
    _to_java_object_rdd(rdd)
))

# Reserialize and convert to JavaDStream<Object>
# This is the only option which allows further transformations
# on DStream
ssc._jvm.dummy.PythonDStreamHelper.go(
    q.transform(lambda rdd: RDD(  # Reserialize but keep as Python RDD
        _to_java_object_rdd(rdd), ssc.sparkContext
    ))._jdstream
)

# Convert to DataFrame and pass to Scala sink.
# Arguably there are relatively few moving parts here. 
q.foreachRDD(lambda rdd: 
    ssc._jvm.dummy.PythonDataFrameHelper.go(
        rdd.map(lambda x: (x, )).toDF()._jdf
    )
)

ssc.start()
ssc.awaitTerminationOrTimeout(30)
ssc.stop()

这不受支持，未经测试，因此除了 Spark API 实验之外，对其他任何事情都毫无用处。