[BEAM-2852] Nexmark Kafka source sink #3937

vectorijk · 2017-10-03T21:55:40Z

Add support for Kafka as source/sink on Nexmark

still experiment sourceFromKafka

vectorijk · 2017-10-03T21:58:51Z

echauchot · 2017-10-04T08:05:18Z

Thanks for your help @vectorijk ! I'll take a look as soon as I'm done with my current PR.

iemejia

Awesome job, thanks a lot for working on it, I commented on some small fixes. Can you also please add some documentation on how to use Nexmark with Kafka in the Nexmark README. (We are moving this eventually to the web site but we will do this later).

iemejia · 2017-10-06T11:33:39Z

sdks/java/nexmark/pom.xml

+      <groupId>org.apache.beam</groupId>
+      <artifactId>beam-sdks-java-io-kafka</artifactId>
+      <exclusions>
+        <exclusion>


I think this exclusion is not needed. Can you remove it if so.

iemejia · 2017-10-06T12:54:00Z

sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/NexmarkLauncher.java

      }));
  }

+  static final ParDo.SingleOutput<Event, byte[]> EVENT_TO_BYTEARRAY =


private, maybe make the return type DoFn instead of ParDo and apply the ParDo in the other part to make it more Beam style, (and yes I know nexmark does not do this in other parts shame on us :P).

iemejia · 2017-10-06T13:26:25Z

sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/NexmarkLauncher.java

+  }
+
+
+  static final ParDo.SingleOutput<KV<Long, byte[]>, Event> BYTEARRAY_TO_EVENT =


Same as with the other ParDo (private + Fn).

iemejia · 2017-10-06T14:39:11Z

sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/NexmarkLauncher.java

+   * Send {@code events} to Kafka.
+   */
+  private void sinkEventsToKafka(PCollection<Event> events) {
+    if (Strings.isNullOrEmpty(options.getBootstrapServers())) {


Change for Preconditions.checkArgument

It is actually unneeded because this check is already done in KafkaIO.Write.validate. You can remove it.

iemejia · 2017-10-06T14:39:26Z

sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/NexmarkLauncher.java

+  /**
+   * Send {@code formattedResults} to Kafka.
+   */
+  private void sinkResultsToKafka(PCollection<String> formattedResults) {


Change for Preconditions.checkArgument

iemejia · 2017-10-06T14:46:05Z

sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/NexmarkLauncher.java

+  /**
+   * Send {@code events} to Kafka.
+   */
+  private void sinkEventsToKafka(PCollection<Event> events) {


This is currently unused, can you add something similar to Pubsub's PUBLISH_ONLY so users can generate and record the events in Kafka.

IMHO, it might be useful as well to inject events to the benchmark using kafka similarly to what is done in Pusb/Sub COMBINED mode.

vectorijk · 2017-10-09T01:57:55Z

retest this please

echauchot

@vectorijk, thanks for your work!
Besides I have tested your PR by launching query1 in batch mode in direct runner with a kafka sink to a local kafka instance through this command
mvn -Pdirect-runner exec:java -Dexec.mainClass=org.apache.beam.sdk.nexmark.Main "-Dexec.args=--sinkType=KAFKA --bootstrapServers=127.0.0.1:9092 --runner=DirectRunner --query=1 --streaming=false --manageResources=false --monitorJobs=true --enforceEncodability=true --enforceImmutability=true"
And I got this error:

GRAVE: Uncaught error in kafka producer I/O thread: 
org.apache.kafka.common.protocol.types.SchemaException: Error reading field 'topic_metadata': Error reading array of size 1702391137, only 37 bytes available
	at org.apache.kafka.common.protocol.types.Schema.read(Schema.java:73)
	at org.apache.kafka.clients.NetworkClient.parseResponse(NetworkClient.java:380)
	at org.apache.kafka.clients.NetworkClient.handleCompletedReceives(NetworkClient.java:449)
	at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:269)
	at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:236)
	at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:135)
	at java.lang.Thread.run(Thread.java:745)

Maybe related to the fact that you use PCollection<String> formattedResults and apply to it a KafkaIO.<Long, String> with keySerializer whereas there is no key in the PCollection

echauchot · 2017-10-10T16:45:10Z

sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/NexmarkLauncher.java

      }));
  }

+  static final ParDo.SingleOutput<Event, byte[]> EVENT_TO_BYTEARRAY =


echauchot · 2017-10-11T08:49:01Z

sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/NexmarkLauncher.java

+  /**
+   * Send {@code events} to Kafka.
+   */
+  private void sinkEventsToKafka(PCollection<Event> events) {


IMHO, it might be useful as well to inject events to the benchmark using kafka similarly to what is done in Pusb/Sub COMBINED mode.

echauchot · 2017-10-11T15:55:58Z

sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/NexmarkLauncher.java

+   * Send {@code events} to Kafka.
+   */
+  private void sinkEventsToKafka(PCollection<Event> events) {
+    if (Strings.isNullOrEmpty(options.getBootstrapServers())) {


It is actually unneeded because this check is already done in KafkaIO.Write.validate. You can remove it.

vectorijk

Thanks @echauchot @iemejia for the review! I will address comments asap.

vectorijk · 2017-10-12T20:18:26Z

sdks/java/nexmark/pom.xml

+      <groupId>org.apache.beam</groupId>
+      <artifactId>beam-sdks-java-io-kafka</artifactId>
+      <exclusions>
+        <exclusion>


echauchot · 2017-10-31T16:55:54Z

@vectorijk Do you need any help?

echauchot · 2017-11-08T10:19:23Z

ping?

vectorijk · 2017-11-09T10:56:30Z

I'll update this today and tomorrow

…

On Wed, Nov 8, 2017, 02:19 Etienne Chauchot ***@***.***> wrote: ping? — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#3937 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/ADQu6cjoBvLxDMjeGBs9cJHu5suimedLks5s0YAvgaJpZM4Ps1EC> .

vectorijk · 2017-11-10T06:36:03Z

@echauchot I tried mvn clean install -DskipTests exec:java -Dexec.mainClass=org.apache.beam.sdk.nexmark.Main -Pdirect-runner -Dexec.args="--runner=DirectRunner --suite=SMOKE --streaming=false --query=1 --manageResources=true --monitorJobs=true --enforceEncodability=true --enforceImmutability=true --sinkType=KAFKA --bootstrapServers=localhost:9092"

It works for me on my local. I can't reproduce. @iemejia Could you test this on your side?

echauchot · 2017-11-10T09:07:06Z

@vectorijk thanks for the fixes! I still have 2 comments:

what about implementing COMBINED mode?
a remark for the future: you should rebase your branch on master and not merge master into your branch (see beam contribution guide). Can you remove the merge commits (using rebase -i) and, rebase on master?
Waiting for @iemejia to test on his local setup for the org.apache.kafka.common.protocol.types.SchemaException: Error reading field 'topic_metadata'

echauchot

waiting for status on error to see if it is related to my setup and also the COMBINED state.

vectorijk · 2017-11-13T05:56:06Z

@echauchot I will take a look at COMBINED mode and update it soon.

aaltay · 2017-11-28T23:27:43Z

@vectorijk any updates?

kennknowles · 2018-01-26T16:35:03Z

How is this going? This seems nice to have. It is probably a good time to rebase and restart review since Nexmark has change a little and the codebase is Java 8.

vectorijk · 2018-01-26T23:09:43Z

@kennknowles I will change and rebase this.

echauchot · 2018-01-29T09:03:47Z

@vectorijk I'm waiting for the status of the test on @iemejia's setup so that we could see if the org.apache.kafka.common.protocol.types.SchemaException is related to my setup. When it is ok I'll merge the PR. Besides, what about the combined mode?

aromanenko-dev · 2018-04-04T10:15:47Z

I checked out this PR and rebased it against master (there were some minor conflicts, easy to resolve).
Then I run Nexmark as @echauchot suggested above and it works for me:
$ mvn -Pdirect-runner exec:java -Dexec.mainClass=org.apache.beam.sdk.nexmark.Main "-Dexec.args=--sinkType=KAFKA --bootstrapServers=localhost:9092 --runner=DirectRunner --query=1 --streaming=false --manageResources=false --monitorJobs=true --enforceEncodability=true --enforceImmutability=true"

The output result:

Configurations:
  Conf  Description
  0000  query:1; sinkType:KAFKA

Performance:
  Conf  Runtime(sec)    (Baseline)  Events(/sec)    (Baseline)       Results    (Baseline)
  0000          10.5                      9510.2                       92000

echauchot · 2018-04-04T11:59:55Z

@aromanenko-dev thanks for the feedback! So the issue that I reported was due to my setup then. I'll rebase and merge.

echauchot · 2018-04-04T12:00:58Z

There was a point in my review about the COMBINED mode but we could support it later in another PR.

echauchot · 2018-04-04T12:23:31Z

@vectorijk seems to be unavailable so I propose to do another PR for minor fixes: squash (with obviously same author), rebase on master, and wire up unused sinkEventsToKafka method. Then merge that one.

vectorijk added 3 commits October 3, 2017 10:57

sinkToKafka works

46e380c

still experiment sourceFromKafka

source event from kafka

18128d2

clean code

d603dcd

iemejia requested changes Oct 6, 2017

View reviewed changes

echauchot requested changes Oct 12, 2017

View reviewed changes

address comments

d38d956

vectorijk commented Oct 18, 2017

View reviewed changes

vectorijk added 3 commits November 9, 2017 14:52

address comments 2

f2ad04f

Merge branch 'master' into BEAM-2852-nexmark-kafka-source-sink

0d2c63d

Merge branch 'master' into BEAM-2852-nexmark-kafka-source-sink

6df6e17

changes

d246c41

echauchot reviewed Nov 10, 2017

View reviewed changes

echauchot approved these changes Apr 4, 2018

View reviewed changes

echauchot closed this Apr 4, 2018

aromanenko-dev mentioned this pull request Apr 4, 2018

[BEAM-2852] Add support for Kafka as source/sink on Nexmark #5019

Merged

10 tasks

		}


		static final ParDo.SingleOutput<KV<Long, byte[]>, Event> BYTEARRAY_TO_EVENT =

[BEAM-2852] Nexmark Kafka source sink #3937

[BEAM-2852] Nexmark Kafka source sink #3937

Uh oh!

Conversation

vectorijk commented Oct 3, 2017

Uh oh!

vectorijk commented Oct 3, 2017

Uh oh!

echauchot commented Oct 4, 2017

Uh oh!

iemejia left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

vectorijk commented Oct 9, 2017

Uh oh!

echauchot left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

vectorijk left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

echauchot commented Oct 31, 2017

Uh oh!

echauchot commented Nov 8, 2017

Uh oh!

vectorijk commented Nov 9, 2017 via email

Uh oh!

vectorijk commented Nov 10, 2017

Uh oh!

echauchot commented Nov 10, 2017

Uh oh!

echauchot left a comment

Choose a reason for hiding this comment

Uh oh!

vectorijk commented Nov 13, 2017

Uh oh!

aaltay commented Nov 28, 2017

Uh oh!

kennknowles commented Jan 26, 2018

Uh oh!

vectorijk commented Jan 26, 2018

Uh oh!

echauchot commented Jan 29, 2018

Uh oh!

echauchot commented Apr 4, 2018 •

edited

Loading