Mailing-List: contact user-help@spark.apache.org; run by ezmlm
Precedence: bulk
MIME-Version: 1.0
From: Alexis Seigneurin <aseigneurin@ippon.fr>
Date: Mon, 12 Sep 2016 16:47:40 -0400
Message-ID: <CA+LY3q=1fjKuF64fq8OOS2bQHTqDhf34QdMxLApwEW_cGVrUyw@mail.gmail.com>
Subject: Strings not converted when calling Scala code from a PySpark app
To: user <user@spark.apache.org>
Content-Type: multipart/alternative; boundary=001a1143ce206723c5053c559baf
archived-at: Mon, 12 Sep 2016 20:47:56 -0000

--001a1143ce206723c5053c559baf
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: quoted-printable

Hi,


*TL;DR - I have what looks like a DStream of Strings in a PySpark
application. I want to send it as a DStream[String] to a Scala library.
Strings are not converted by Py4j, though.*


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 DStream[String] 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.

My question would be: how to get Java strings out of the DStream object?


Here is the simplest Python code I came up with:

from pyspark.streaming import StreamingContext
ssc =3D StreamingContext(sparkContext=3Dsc, batchDuration=3Dint(1))

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

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

ssc.start()


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

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


On the Scala side, I have:

package com.seigneurin

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

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


Now, let's say I send some data into Kafka:

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


The println statement in the Scala code prints something that looks like:

[B@758aa4d9


I expected to get foo bar instead.

Now, if I replace the simple println statement in the Scala code with the
following:

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


I get:

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


This suggests that the strings are actually passed as arrays of bytes.

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

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


I get something that looks like (special characters might be stripped off):

=EF=BF=BD]qXfoo barqa.


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


Thanks,
Alexis

--001a1143ce206723c5053c559baf
Content-Type: text/html; charset=UTF-8
Content-Transfer-Encoding: quoted-printable

<div dir=3D"ltr"><div>Hi,</div><div><b><br></b></div><div><b><br></b></div>=
<div><b>TL;DR - I have what looks like a DStream of Strings in a PySpark ap=
plication. I want to send it as a DStream[String] to a Scala library. Strin=
gs are not converted by Py4j, though.</b></div><div><br></div><div><br></di=
v><div>I&#39;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 m=
ethod in Scala code, passing it a DStream[String] instance. However, I&#39;=
m unable to receive proper JVM strings in the Scala code. It looks to me li=
ke the Python strings are not converted into Java strings but, instead, are=
 serialized.</div><div><br></div><div>My question would be: how to get Java=
 strings out of the DStream object?</div><div><br></div><div><br></div><div=
>Here is the simplest Python code I came up with:</div><div><br></div><bloc=
kquote style=3D"margin:0 0 0 40px;border:none;padding:0px"><div><font face=
=3D"monospace, monospace">from pyspark.streaming import StreamingContext</f=
ont></div><div><font face=3D"monospace, monospace">ssc =3D StreamingContext=
(sparkContext=3Dsc, batchDuration=3Dint(1))</font></div><div><font face=3D"=
monospace, monospace"><br></font></div><div><font face=3D"monospace, monosp=
ace">from pyspark.streaming.kafka import KafkaUtils</font></div><div><font =
face=3D"monospace, monospace">stream =3D KafkaUtils.createDirectStream(ssc,=
 [&quot;IN&quot;], {&quot;metadata.broker.list&quot;: &quot;localhost:9092&=
quot;})</font></div><div><font face=3D"monospace, monospace">values =3D str=
eam.map(lambda tuple: tuple[1])</font></div><div><font face=3D"monospace, m=
onospace"><br></font></div><div><font face=3D"monospace, monospace">ssc._jv=
m.com.seigneurin.MyPythonHelper.doSomething(values._jdstream)</font></div><=
div><font face=3D"monospace, monospace"><br></font></div><div><font face=3D=
"monospace, monospace">ssc.start()</font></div></blockquote><div><br></div>=
<div>I&#39;m running this code in PySpark, passing it the path to my JAR:</=
div><div><br></div><blockquote style=3D"margin:0 0 0 40px;border:none;paddi=
ng:0px"><div><font face=3D"monospace, monospace">pyspark --driver-class-pat=
h ~/path/to/my/lib-0.1.1-SNAPSHOT.jar</font></div></blockquote><div><br></d=
iv><div>On the Scala side, I have:</div><div><br></div><blockquote style=3D=
"margin:0 0 0 40px;border:none;padding:0px"><div><font face=3D"monospace, m=
onospace">package com.seigneurin</font></div><div><font face=3D"monospace, =
monospace"><br></font></div><div><font face=3D"monospace, monospace">import=
 org.apache.spark.streaming.api.java.JavaDStream</font></div><div><font fac=
e=3D"monospace, monospace"><br></font></div><div><font face=3D"monospace, m=
onospace">object MyPythonHelper {</font></div><div><font face=3D"monospace,=
 monospace">=C2=A0 def doSomething(jdstream: JavaDStream[String]) =3D {</fo=
nt></div><div><font face=3D"monospace, monospace">=C2=A0 =C2=A0 val dstream=
 =3D jdstream.dstream</font></div><div><font face=3D"monospace, monospace">=
=C2=A0 =C2=A0 dstream.foreachRDD(rdd =3D&gt; {</font></div><div><font face=
=3D"monospace, monospace">=C2=A0 =C2=A0 =C2=A0 rdd.foreach(println)</font><=
/div><div><font face=3D"monospace, monospace">=C2=A0 =C2=A0 })</font></div>=
<div><font face=3D"monospace, monospace">=C2=A0 }</font></div><div><font fa=
ce=3D"monospace, monospace">}</font></div></blockquote><div><br></div><div>=
Now, let&#39;s say I send some data into Kafka:</div><div><br></div><blockq=
uote style=3D"margin:0 0 0 40px;border:none;padding:0px"><div><font face=3D=
"monospace, monospace">echo &#39;foo bar&#39; | $KAFKA_HOME/bin/kafka-conso=
le-producer.sh --broker-list localhost:9092 --topic IN</font></div></blockq=
uote><div><br></div><div>The println statement in the Scala code prints som=
ething that looks like:</div><div><br></div><blockquote style=3D"margin:0 0=
 0 40px;border:none;padding:0px"><div><font face=3D"monospace, monospace">[=
B@758aa4d9</font></div></blockquote><div><br></div><div>I expected to get f=
oo bar instead.</div><div><br></div><div>Now, if I replace the simple print=
ln statement in the Scala code with the following:</div><div><br></div><blo=
ckquote style=3D"margin:0 0 0 40px;border:none;padding:0px"><div><font face=
=3D"monospace, monospace">rdd.foreach(v =3D&gt; println(v.getClass.getCanon=
icalName))</font></div></blockquote><div><br></div><div>I get:</div><div><b=
r></div><blockquote style=3D"margin:0 0 0 40px;border:none;padding:0px"><di=
v><font face=3D"monospace, monospace">java.lang.ClassCastException: [B cann=
ot be cast to java.lang.String</font></div></blockquote><div><br></div><div=
>This suggests that the strings are actually passed as arrays of bytes.</di=
v><div><br></div><div>If I simply try to convert this array of bytes into a=
 string (I know I&#39;m not even specifying the encoding):</div><div><br></=
div><blockquote style=3D"margin:0 0 0 40px;border:none;padding:0px"><div><f=
ont face=3D"monospace, monospace">=C2=A0 =C2=A0 =C2=A0 def doSomething(jdst=
ream: JavaDStream[Array[Byte]]) =3D {</font></div><div><font face=3D"monosp=
ace, monospace">=C2=A0 =C2=A0 =C2=A0 =C2=A0 val dstream =3D jdstream.dstrea=
m</font></div><div><font face=3D"monospace, monospace">=C2=A0 =C2=A0 =C2=A0=
 =C2=A0 dstream.foreachRDD(rdd =3D&gt; {</font></div><div><font face=3D"mon=
ospace, monospace">=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 rdd.foreach(bytes =3D=
&gt; println(new String(bytes)))</font></div><div><font face=3D"monospace, =
monospace">=C2=A0 =C2=A0 =C2=A0 =C2=A0 })</font></div><div><font face=3D"mo=
nospace, monospace">=C2=A0 =C2=A0 =C2=A0 }</font></div></blockquote><div><b=
r></div><div>I get something that looks like (special characters might be s=
tripped off):</div><div><br></div><blockquote style=3D"margin:0 0 0 40px;bo=
rder:none;padding:0px"><div><font face=3D"monospace, monospace">=EF=BF=BD]q=
Xfoo barqa.</font></div></blockquote><div><br></div><div>This suggests the =
Python string was serialized (pickled?). How could I retrieve a proper Java=
 string instead?</div><div><br></div><div><br></div><div>Thanks,</div><div>=
Alexis</div></div>

--001a1143ce206723c5053c559baf--