Transforming PySpark RDD with Scala
<p><strong>TL;DR - I have what looks like a DStream of Strings in a PySpark application. I want to send it as a <code>DStream[String]</code> to a Scala library. Strings are not converted by Py4j, though.</strong></p>

<p>I'm working on a PySpark application that pulls data from Kafka using Spark Streaming. My messages are strings and I would like to call a method in Scala code, passing it a <code>DStream[String]</code> instance. However, I'm unable to receive proper JVM strings in the Scala code. It looks to me like the Python strings are not converted into Java strings but, instead, are serialized.</p>

<p>My question would be: how to get Java strings out of the <code>DStream</code> object?</p>

<hr>

<p>Here is the simplest Python code I came up with:</p>

<pre><code>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()
</code></pre>

<p>I'm running this code in PySpark, passing it the path to my JAR:</p>

<pre><code>pyspark --driver-class-path ~/path/to/my/lib-0.1.1-SNAPSHOT.jar
</code></pre>

<p>On the Scala side, I have:</p>

<pre><code>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 =&gt; {
      rdd.foreach(println)
    })
  }
}
</code></pre>

<p>Now, let's say I send some data into Kafka:</p>

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

<p>The <code>println</code> statement in the Scala code prints something that looks like:</p>

<pre><code>[B@758aa4d9
</code></pre>

<p>I expected to get <code>foo bar</code> instead.</p>

<p>Now, if I replace the simple <code>println</code> statement in the Scala code with the following:</p>

<pre><code>rdd.foreach(v =&gt; println(v.getClass.getCanonicalName))
</code></pre>

<p>I get:</p>

<pre><code>java.lang.ClassCastException: [B cannot be cast to java.lang.String
</code></pre>

<p>This suggests that the strings are actually passed as arrays of bytes.</p>

<p>If I simply try to convert this array of bytes into a string (I know I'm not even specifying the encoding):</p>

<pre><code>      def doSomething(jdstream: JavaDStream[Array[Byte]]) = {
        val dstream = jdstream.dstream
        dstream.foreachRDD(rdd =&gt; {
          rdd.foreach(bytes =&gt; println(new String(bytes)))
        })
      }
</code></pre>

<p>I get something that <em>looks</em> like (special characters might be stripped off):</p>

<pre><code>�]qXfoo barqa.
</code></pre>

<p>This suggests the Python string was serialized (pickled?). How could I retrieve a proper Java string instead?</p>