Added spark streaming custom receiver for pulsar #296

yush1ga · 2017-03-17T08:19:29Z

Motivation

Recent years, more and more people are interested in Apache Spark for machine learning or something and there is demand for flowing data from message queue into Apache Spark.

Modifications

Added Spark Streaming Custom Receiver for Pulsar and its test.

Result

Pulsar can be used as a data source for Apache Spark.

saandrews · 2017-03-20T23:01:27Z

pulsar-spark/src/main/java/com/yahoo/pulsar/spark/SparkStreamingPulsarReceiver.java

+            pulsarClient = PulsarClient.create(url, clientConfiguration);
+            consumerConfiguration.setMessageListener(new MessageListener() {
+                public void received(Consumer consumer, Message msg) {
+                    store(new String(msg.getData()));


Can we avoid the copy and pass the buffer instead?

Does store() throw any exception? If so, we should also handle it right?

I think if we change from extends Receiver<String> to extends Receiver<byte[]>, we will be able to avoid the copy.
Does it look good ?

saandrews · 2017-03-20T23:12:17Z

pulsar-spark/src/main/java/com/yahoo/pulsar/spark/SparkStreamingPulsarReceiver.java

+    }
+
+    private static final Logger log = LoggerFactory.getLogger(SparkStreamingPulsarReceiver.class);
+}


We should also update documentation .

OK. I'll make it.

saandrews · 2017-03-20T23:13:39Z

pulsar-spark/src/main/java/com/yahoo/pulsar/spark/SparkStreamingPulsarReceiver.java

+            });
+            pulsarClient.subscribe(topic, subscription, consumerConfiguration);
+        } catch (PulsarClientException e) {
+            restart(e.getMessage(), e);


The exception is logged by restart()? Else it's better to log it here.

there are situations where client failures can be seen for a reasonably long period of time, so you might want to think about resource consumption in that case given that restart() will recurse into onStart().

you create a new PulsarClient on each call, turning the previous one into garbage.

you create a new MessageListener implementation on each call, with similar garbage.

i'd set those up in the constructor while leaving the call to subscribe() to onStart().

@saandrews
I added logging process.

@msb-at-yahoo
I reimplemented like that.

msb-at-yahoo · 2017-03-21T05:55:05Z

pulsar-spark/src/main/java/com/yahoo/pulsar/spark/SparkStreamingPulsarReceiver.java

+
+    public void onStop() {
+        try {
+            if (pulsarClient != null) {


is this transition valid? onStart -> onStop -> onStart?

onStart and onStop are called only once.
onStart -> onStop

ok. the reason i asked is that it's now a bit asymmetric after you made the change i suggested to onStart(): onStop() invalidates the invariants that onStart() relies upon.

Sorry, I confirmed that when restart() is called, onStop() -> onStart() is called.
As we create a client in the constructor, it cannot start subscribing when restarting.
I think we should create a client in onStart().

yush1ga · 2017-03-21T07:00:51Z

I added documentation.

merlimat · 2017-03-21T19:31:24Z

pulsar-spark/src/main/java/com/yahoo/pulsar/spark/SparkStreamingPulsarReceiver.java

+    private String topic;
+    private String subscription;
+
+    public SparkStreamingPulsarReceiver(


Please format the files with the Eclipse formatter profile that can be found at src/formatter.xml

I applied the format.

merlimat · 2017-03-21T19:33:50Z

pulsar-spark/src/main/java/com/yahoo/pulsar/spark/SparkStreamingPulsarReceiver.java

+        this.topic = topic;
+        this.subscription = subscription;
+        consumerConfiguration.setMessageListener((consumer, msg) -> {
+            store(new String(msg.getData()));


What happens if store() fails?

If you want to retry the operation, you should then block the listener thread until the store() succeeds.

I couldn't find the case that store() fails in the official documents and others.
I think spark streaming is mainly used for calculating statistical data and so on and we don't have to mind some data lost in those use cases.
What do you think?

I think there are 2 options in case the store() calls fails with exception:

Don't ack the message and rely on ack-timeout to replay the message some time later (eg: 1min)

Just ack the message anyway and move on, basically ignoring the exception.

I prefer option 1 by default.

@merlimat

I chose option 1.

AckTimeout is set if it's not set

Added try block

When failing to store, consumer doesn't ack

The message will be resent based on AckTimeout in consumerConfiguration.

merlimat · 2017-03-21T19:34:44Z

pulsar-spark/src/main/java/com/yahoo/pulsar/spark/SparkStreamingPulsarReceiver.java

+            pulsarClient.subscribe(topic, subscription, consumerConfiguration);
+        } catch (PulsarClientException e) {
+            log.error("Failed to start subscription : {}", e.getMessage());
+            restart("Restart a consumer");


Wouldn't be possible to just propagate the exception at this point?

PulsarClientException is so a checked exception that it has to be caught here.
If we try to propagate like following, onStart() cannot be overrode.

public void onStart() throws PulsarClientExceptio{

catch (PulsarClientException e) { throw new PulsarClientException("error!"); }

msb-at-yahoo · 2017-03-21T19:39:19Z

pulsar-spark/src/main/java/com/yahoo/pulsar/spark/SparkStreamingPulsarReceiver.java

+        this.topic = topic;
+        this.subscription = subscription;
+        consumerConfiguration.setMessageListener((consumer, msg) -> {
+            store(new String(msg.getData()));


besides the concerns from @saandrews regarding the data copy, by using String will this prevent Spark users from consuming binary messages?

Maybe, that is right.
CustomReceiver may have to extend Receiver<byte[]> to make users be able consume binary messages.
I will implement like so.

merlimat

👍

Fix apache#290 This PR supports continuous offset for kop. Since this PR disable the orginal design of mapping between Pulsar `MessageId` and Kafka `offset`. So I just ignore some UnitTest based on the original desing. Maybe we can raise up another issue to track how we deal with these ignored tests.

Fixes apache#312 These tests were ignored temporarily just because they rely on the outdated methods that convert between MessageId and Kafka offset. So this PR fixes these tests and deletes these outdated methods. The exception is testBrokerRespectsPartitionsOrderAndSizeLimits, it's a broken test that is easily affected by some params. apache#287 and apache#246 are tracking the test.

saandrews reviewed Mar 20, 2017

View reviewed changes

yush1ga force-pushed the pulsar-spark branch from 2d9f4eb to f1b2e72 Compare March 21, 2017 05:04

msb-at-yahoo reviewed Mar 21, 2017

View reviewed changes

yush1ga force-pushed the pulsar-spark branch from f1b2e72 to 0d7a609 Compare March 21, 2017 07:00

merlimat reviewed Mar 21, 2017

View reviewed changes

msb-at-yahoo reviewed Mar 21, 2017

View reviewed changes

merlimat added this to the 1.17 milestone Mar 21, 2017

merlimat added the type/feature The PR added a new feature or issue requested a new feature label Mar 21, 2017

yush1ga force-pushed the pulsar-spark branch 6 times, most recently from 5b6879c to 547fb8a Compare March 28, 2017 05:01

Added spark streaming custom receiver for pulsar

181d19a

yush1ga force-pushed the pulsar-spark branch from 547fb8a to 181d19a Compare March 28, 2017 05:50

merlimat approved these changes Mar 30, 2017

View reviewed changes

merlimat modified the milestones: 1.17, 1.18 Mar 31, 2017

Merge branch 'master' into pulsar-spark

63e1b68

merlimat merged commit b1df955 into apache:master Apr 4, 2017

merlimat mentioned this pull request Apr 4, 2017

Fixed pulsar-spark pom version #330

Merged

yush1ga deleted the pulsar-spark branch April 18, 2017 04:13

xiaotongwang1 mentioned this pull request Aug 4, 2021

Pulsar 2.7.0+ KOP 2.7.2.x getPartitionedTopicMetadata timeout #11532

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Added spark streaming custom receiver for pulsar #296

Added spark streaming custom receiver for pulsar #296

yush1ga commented Mar 17, 2017

saandrews Mar 20, 2017

yush1ga Mar 21, 2017

saandrews Mar 20, 2017

yush1ga Mar 21, 2017

saandrews Mar 20, 2017

msb-at-yahoo Mar 20, 2017

yush1ga Mar 21, 2017

msb-at-yahoo Mar 21, 2017

yush1ga Mar 21, 2017

msb-at-yahoo Mar 21, 2017

yush1ga Mar 22, 2017 •

edited

Loading

yush1ga commented Mar 21, 2017

merlimat Mar 21, 2017

yush1ga Mar 22, 2017

merlimat Mar 21, 2017

yush1ga Mar 22, 2017

merlimat Mar 24, 2017

yush1ga Mar 27, 2017

merlimat Mar 21, 2017

yush1ga Mar 22, 2017

msb-at-yahoo Mar 21, 2017

yush1ga Mar 22, 2017

merlimat left a comment

Added spark streaming custom receiver for pulsar #296

Added spark streaming custom receiver for pulsar #296

Conversation

yush1ga commented Mar 17, 2017

Motivation

Modifications

Result

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

yush1ga Mar 22, 2017 • edited Loading

Choose a reason for hiding this comment

yush1ga commented Mar 21, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

merlimat left a comment

Choose a reason for hiding this comment

yush1ga Mar 22, 2017 •

edited

Loading