[SPARK-9065][Streaming][PySpark] Add MessageHandler for Kafka Python API #7410

jerryshao · 2015-07-15T03:12:20Z

Propose a way to add messageHandler for Kafka Python API, offer user a similar pattern to use message like Scala/Java:

def getOffset(m):
    return m and m.offset

KafkaUtils.createDirectStream(ssc, topic, kafkaParams, messageHandler=getOffset)

Internally serialize the Kafka MessageAndMetadata to Python using Pickler.

Please help to review, thanks a lot.

SparkQA · 2015-07-15T03:44:32Z

Test build #37305 has finished for PR 7410 at commit 798b047.

This patch fails MiMa tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class PythonMessageAndMetadata(
- class PythonMessageAndMetadataPickler extends IObjectPickler
- class KafkaMessageAndMetadata(object):
- case class Exchange(newPartitioning: Partitioning, child: SparkPlan) extends UnaryNode

SparkQA · 2015-07-15T07:38:32Z

Test build #37314 has finished for PR 7410 at commit 0eabce4.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class PythonMessageAndMetadata(
- class PythonMessageAndMetadataPickler extends IObjectPickler
- class KafkaMessageAndMetadata(object):

SparkQA · 2015-07-15T11:35:11Z

Test build #37348 has finished for PR 7410 at commit f23d1cb.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class PythonMessageAndMetadata(
- class PythonMessageAndMetadataPickler extends IObjectPickler
- class KafkaMessageAndMetadata(object):
- case class Exchange(newPartitioning: Partitioning, child: SparkPlan) extends UnaryNode

SparkQA · 2015-07-15T17:52:35Z

Test build #1072 has finished for PR 7410 at commit f23d1cb.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

zsxwing · 2015-07-22T17:15:56Z

@jerryshao could you resolve the conflicts? Thanks.

jerryshao · 2015-07-23T04:36:49Z

Yeah, will do.

SparkQA · 2015-07-24T04:36:53Z

Test build #38295 has finished for PR 7410 at commit a09423e.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class PythonMessageAndMetadata(
- class PythonMessageAndMetadataPickler extends IObjectPickler
- class KafkaMessageAndMetadata(object):
- case class ChangePrecision(child: Expression) extends UnaryExpression
- case class DecimalType(precision: Int, scale: Int) extends FractionalType
- case class DecimalConversion(precision: Int, scale: Int) extends JDBCConversion

jerryshao · 2015-07-24T04:53:39Z

Jenkins, retest this please.

SparkQA · 2015-07-24T06:51:53Z

Test build #38309 has finished for PR 7410 at commit a09423e.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class PythonMessageAndMetadata(
- class PythonMessageAndMetadataPickler extends IObjectPickler
- class KafkaMessageAndMetadata(object):
- case class ChangePrecision(child: Expression) extends UnaryExpression
- abstract class AlgebraicAggregate extends AggregateFunction2 with Serializable with Unevaluable
- abstract class AggregateFunction1 extends LeafExpression with Serializable
- case class DecimalType(precision: Int, scale: Int) extends FractionalType
- case class DecimalConversion(precision: Int, scale: Int) extends JDBCConversion

SparkQA · 2015-07-24T07:08:03Z

Test build #85 has finished for PR 7410 at commit a09423e.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class PythonMessageAndMetadata(
- class PythonMessageAndMetadataPickler extends IObjectPickler
- class KafkaMessageAndMetadata(object):
- case class ChangePrecision(child: Expression) extends UnaryExpression
- abstract class AlgebraicAggregate extends AggregateFunction2 with Serializable with Unevaluable
- abstract class AggregateFunction1 extends LeafExpression with Serializable
- case class DecimalType(precision: Int, scale: Int) extends FractionalType
- case class DecimalConversion(precision: Int, scale: Int) extends JDBCConversion

zsxwing · 2015-07-27T15:43:12Z

external/kafka/src/main/scala/org/apache/spark/streaming/kafka/KafkaUtils.scala

+private object KafkaUtilsPythonHelper {
+  private var initialized = false
+
+  def initialize(): Unit = {


Why not directly call SerDeUtil.initialize() and new PythonMessageAndMetadataPickler().register() in the static codes of KafkaUtilsPythonHelper? It would be much simpler since it's only two lines.

Oh, I see. You want to call initialize to load the object KafkaUtilsPythonHelper in the closure. If so, how about adding a method in KafkaUtilsPythonHelper, such as

object KafkaUtilsPythonHelper { ... def createPicklerIterator: Iterator[Array[Byte]] = { new SerDeUtil.AutoBatchedPickler(iter) } ... }

and call it in the closure?

Would you please share more details on it, I cannot exactly get what you mean. What I did here is to follow the pattern here in MLlib(https://github.com/apache/spark/blob/master/mllib/src/main/scala/org/apache/spark/mllib/api/python/PythonMLLibAPI.scala#L1370)

Since initialize() is already called in object KafkaUtilsPythonHelper, the purpose of adding the initialize method is only for loading the class of object KafkaUtilsPythonHelper, right?

If changing it to

object KafkaUtilsPythonHelper { SerDeUtil.initialize() new PythonMessageAndMetadataPickler().register() ... def createPicklerIterator: Iterator[Array[Byte]] = { new SerDeUtil.AutoBatchedPickler(iter) } ... }

When calling KafkaUtilsPythonHelper.createPicklerIterator in the closure, SerDeUtil.initialize() and new PythonMessageAndMetadataPickler().register() will be called when loading class of object KafkaUtilsPythonHelper.

SparkQA · 2015-07-28T07:43:15Z

Test build #38661 has finished for PR 7410 at commit 348e3a2.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class PythonMessageAndMetadata(
- class PythonMessageAndMetadataPickler extends IObjectPickler
- class KafkaMessageAndMetadata(object):

zsxwing · 2015-08-05T01:05:19Z

Sorry for the delay. LGTM except the conflicts. Could you resolve the conflicts again?

jerryshao · 2015-08-05T01:57:25Z

OK, I will rebase the code.

SparkQA · 2015-08-05T05:35:41Z

Test build #39815 has finished for PR 7410 at commit f375e16.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class PythonMessageAndMetadata(
- class PythonMessageAndMetadataPickler extends IObjectPickler
- class KafkaMessageAndMetadata(object):

jerryshao · 2015-08-05T06:49:14Z

Hi @tdas, would you please help to review this patch, thanks a lot.

zsxwing · 2015-08-05T07:24:03Z

LGTM

tdas · 2015-08-06T21:29:52Z

Hey all, this is a pretty good PR. Unfortunately, I havent been able to spend cycles on this as I was busy with higher priority PRs. The Java data being serialized in python does add some overheads, and introduces some data structures, that we need to think a bit. So I think I am going to bump this to Spark 1.6. But this is important feature for python parity so we are definitely going to have this in 1.6.

tdas · 2015-10-07T02:19:39Z

@jerryshao Can you update this PR with latest master?

jerryshao · 2015-10-08T00:36:11Z

Yeah, I will.

SparkQA · 2015-10-08T05:43:47Z

Test build #43380 has finished for PR 7410 at commit 2324517.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class PythonMessageAndMetadata(
- class PythonMessageAndMetadataPickler extends IObjectPickler
- class KafkaMessageAndMetadata(object):

jerryshao · 2015-10-08T05:49:37Z

Jenkins, retest this please.

SparkQA · 2015-10-08T07:39:21Z

Test build #43384 has finished for PR 7410 at commit 2324517.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class PythonMessageAndMetadata(
- class PythonMessageAndMetadataPickler extends IObjectPickler
- class KafkaMessageAndMetadata(object):

SparkQA · 2015-10-10T04:45:41Z

Test build #43512 has finished for PR 7410 at commit a794127.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class PythonMessageAndMetadata(
- class PythonMessageAndMetadataPickler extends IObjectPickler
- class ChildProcAppHandle implements SparkAppHandle
- abstract class LauncherConnection implements Closeable, Runnable
- final class LauncherProtocol
- static class Message implements Serializable
- static class Hello extends Message
- static class SetAppId extends Message
- static class SetState extends Message
- static class Stop extends Message
- class LauncherServer implements Closeable
- class NamedThreadFactory implements ThreadFactory
- class OutputRedirector
- class KafkaMessageAndMetadata(object):

jerryshao · 2015-10-10T06:06:46Z

Jenkins, retest this please.

SparkQA · 2015-10-10T06:29:40Z

Test build #43521 has finished for PR 7410 at commit a794127.

This patch fails MiMa tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class PythonMessageAndMetadata(
- class PythonMessageAndMetadataPickler extends IObjectPickler
- class KafkaMessageAndMetadata(object):

jerryshao · 2015-10-10T06:35:10Z

Jenkins, retest this please.

SparkQA · 2015-10-10T08:38:21Z

Test build #43524 has finished for PR 7410 at commit a794127.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class PythonMessageAndMetadata(
- class PythonMessageAndMetadataPickler extends IObjectPickler
- class KafkaMessageAndMetadata(object):

jerryshao · 2015-10-13T00:52:40Z

@tdas , I've updated the code to the latest master, mind taking a time to review? Thanks a lot.

ontarionick · 2015-11-05T18:37:46Z

Sorry to bump this, but I was wondering if was still the plan to merge this? I'd find this feature quite useful for something I'm building! 😄

zsxwing · 2015-11-05T19:25:09Z

@jerryshao could you update this one again... Thanks!

jerryshao · 2015-11-06T07:10:48Z

Yeah, will do.

SparkQA · 2015-11-06T11:39:08Z

Test build #45216 has finished for PR 7410 at commit ed4d179.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):\n * final class ShuffleSortDataFormat extends SortDataFormat<PackedRecordPointer, LongArray>\n * final class UnsafeSortDataFormat extends SortDataFormat<RecordPointerAndKeyPrefix, LongArray>\n * case class PythonMessageAndMetadata(\n * class PythonMessageAndMetadataPickler extends IObjectPickler\n * class KafkaMessageAndMetadata(object):\n

zsxwing · 2015-11-06T22:34:01Z

LGTM

jerryshao · 2015-11-07T03:51:22Z

Hi @tdas, I understand you're quite busy, still I would like to know your opinion on this PR, I've rebased for several rounds, still pending to review. Thanks a lot and appreciated.

zsxwing · 2015-11-13T17:52:44Z

@jerryshao I just fixed the Streaming Python tests to report test failures correctly #9669 . Could you rebase this one again? Sorry about that.

tdas · 2015-11-16T19:35:46Z

@jerryshao sorry for so much delays on this PR. We needed to fix issues like #9669 before merging python streaming features. Hence the delay. Would you mind if we take over the feature to fix it up in the interest of time. You will get all the credit for this PR on github and JIRA.

jerryshao · 2015-11-17T00:28:44Z

Hi @tdas , it's all right. sorry for the delay, I planned to address this conflict today.

Fixed the merge conflicts in #7410 Closes #7410 Author: Shixiong Zhu <shixiong@databricks.com> Author: jerryshao <saisai.shao@intel.com> Author: jerryshao <sshao@hortonworks.com> Closes #9742 from zsxwing/pr7410. (cherry picked from commit 75a2922) Signed-off-by: Tathagata Das <tathagata.das1565@gmail.com>

jerryshao changed the title ~~[SPARK-8337][Streaming][Pyspark] Add MessageHandler for Kafka Python API~~ [SPARK-8337][Streaming][PySpark] Add MessageHandler for Kafka Python API Jul 15, 2015

jerryshao changed the title ~~[SPARK-8337][Streaming][PySpark] Add MessageHandler for Kafka Python API~~ [SPARK-9065][Streaming][PySpark] Add MessageHandler for Kafka Python API Jul 15, 2015

jerryshao force-pushed the SPARK-8337 branch from f23d1cb to a09423e Compare July 24, 2015 02:39

zsxwing reviewed Jul 27, 2015
View reviewed changes

jerryshao force-pushed the SPARK-8337 branch from 348e3a2 to f375e16 Compare August 5, 2015 03:09

jerryshao force-pushed the SPARK-8337 branch from f375e16 to 2324517 Compare October 8, 2015 04:03

jerryshao force-pushed the SPARK-8337 branch from 2324517 to a794127 Compare October 10, 2015 03:00

jerryshao and others added 5 commits November 6, 2015 16:29

Add MessageHandler for Kafka Python API

5e7a093

Add Mima excludes

e739d33

Fix unit test bug

486ddea

Address the comments

dce508f

Typo fix

ed4d179

jerryshao force-pushed the SPARK-8337 branch from a794127 to ed4d179 Compare November 6, 2015 09:27

zsxwing mentioned this pull request Nov 16, 2015

[SPARK-9065][Streaming][PySpark] Add MessageHandler for Kafka Python API #9742

Closed

asfgit closed this in 75a2922 Nov 18, 2015

[SPARK-9065][Streaming][PySpark] Add MessageHandler for Kafka Python API #7410

[SPARK-9065][Streaming][PySpark] Add MessageHandler for Kafka Python API #7410

Uh oh!

Conversation

jerryshao commented Jul 15, 2015

Uh oh!

SparkQA commented Jul 15, 2015

Uh oh!

SparkQA commented Jul 15, 2015

Uh oh!

SparkQA commented Jul 15, 2015

Uh oh!

SparkQA commented Jul 15, 2015

Uh oh!

zsxwing commented Jul 22, 2015

Uh oh!

jerryshao commented Jul 23, 2015

Uh oh!

SparkQA commented Jul 24, 2015

Uh oh!

jerryshao commented Jul 24, 2015

Uh oh!

SparkQA commented Jul 24, 2015

Uh oh!

SparkQA commented Jul 24, 2015

Uh oh!

zsxwing Jul 27, 2015

Choose a reason for hiding this comment

Uh oh!

zsxwing Jul 27, 2015

Choose a reason for hiding this comment

Uh oh!

jerryshao Jul 28, 2015

Choose a reason for hiding this comment

Uh oh!

zsxwing Jul 28, 2015

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jul 28, 2015

Uh oh!

zsxwing commented Aug 5, 2015

Uh oh!

jerryshao commented Aug 5, 2015

Uh oh!

SparkQA commented Aug 5, 2015

Uh oh!

jerryshao commented Aug 5, 2015

Uh oh!

zsxwing commented Aug 5, 2015

Uh oh!

tdas commented Aug 6, 2015

Uh oh!

tdas commented Oct 7, 2015

Uh oh!

jerryshao commented Oct 8, 2015

Uh oh!

SparkQA commented Oct 8, 2015

Uh oh!

jerryshao commented Oct 8, 2015

Uh oh!

SparkQA commented Oct 8, 2015

Uh oh!

SparkQA commented Oct 10, 2015

Uh oh!

jerryshao commented Oct 10, 2015

Uh oh!

SparkQA commented Oct 10, 2015

Uh oh!

jerryshao commented Oct 10, 2015

Uh oh!

SparkQA commented Oct 10, 2015

Uh oh!

jerryshao commented Oct 13, 2015

Uh oh!

ontarionick commented Nov 5, 2015

Uh oh!

zsxwing commented Nov 5, 2015

Uh oh!

jerryshao commented Nov 6, 2015

Uh oh!