[BEAM-12076] Adds a Kafka external read transform for reading with metadata by chamikaramj · Pull Request #15028 · apache/beam

chamikaramj · 2021-06-17T00:13:56Z

Adds a Kafka external read transform for reading with metadata.
Also updates Python Kafka wrapper to support reading with metadata.

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Choose reviewer(s) and mention them in a comment (R: @username).
Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

`ValidatesRunner` compliance status (on master branch)

Lang	ULR	Samza	Twister2
Go	---	---	---
Java
Python	---	---	---
XLang		---	---

Examples testing status on various runners

Lang	ULR	Dataflow	Flink	Samza	Spark	Twister2
Go	---	---	---	---	---	---	---
Java	---		---	---	---	---	---
Python	---	---	---	---	---	---	---
XLang	---	---	---	---	---	---	---

Post-Commit SDK/Transform Integration Tests Status (on master branch)

Go	Java	Python

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website	Whitespace	Typescript
Non-portable
Portable	---			---	---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

codecov · 2021-06-17T21:32:40Z

Codecov Report

Merging #15028 (7906f96) into master (5fb31eb) will increase coverage by 0.02%.
The diff coverage is 15.38%.

@@            Coverage Diff             @@
##           master   #15028      +/-   ##
==========================================
+ Coverage   83.77%   83.79%   +0.02%     
==========================================
  Files         439      441       +2     
  Lines       59052    59509     +457     
==========================================
+ Hits        49469    49866     +397     
- Misses       9583     9643      +60

Impacted Files	Coverage Δ
...ython/apache_beam/examples/kafkataxi/kafka_taxi.py	`0.00% <0.00%> (ø)`
sdks/python/apache_beam/io/kafka.py	`80.00% <100.00%> (+0.83%)`	⬆️
sdks/python/apache_beam/io/mongodbio.py	`92.88% <0.00%> (-2.76%)`	⬇️
...on/apache_beam/runners/portability/spark_runner.py	`67.34% <0.00%> (-2.22%)`	⬇️
sdks/python/apache_beam/pvalue.py	`91.35% <0.00%> (-1.67%)`	⬇️
...pache_beam/runners/interactive/interactive_beam.py	`74.72% <0.00%> (-1.10%)`	⬇️
sdks/python/apache_beam/internal/metrics/metric.py	`86.17% <0.00%> (-1.07%)`	⬇️
...m/runners/portability/spark_uber_jar_job_server.py	`84.32% <0.00%> (-1.06%)`	⬇️
...ython/apache_beam/io/gcp/bigquery_read_internal.py	`58.46% <0.00%> (-0.75%)`	⬇️
...ks/python/apache_beam/runners/worker/data_plane.py	`90.60% <0.00%> (-0.61%)`	⬇️
... and 26 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 5fb31eb...7906f96. Read the comment docs.

…data

chamikaramj · 2021-06-21T15:25:39Z

R: @ihji

CC: @boyuanzz @TheNeuralBit @youngoli

chamikaramj · 2021-06-23T20:26:46Z

Run Python PreCommit

chamikaramj · 2021-06-24T18:15:59Z

Run Java PreCommit

ihji

Thanks. Looks great overall 😄

Just a few minor comments about naming. They're okay to be ignored if you don't agree.

ihji · 2021-06-25T06:43:02Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java

+    private final Read<K, V> read;
+
+    ExternalWithMetadata(Read<K, V> read) {
+      super("KafkaIO.Read");


TypedWithoutMetadata also sets the same ptransform name KafkaIO.Read:

TypedWithoutMetadata(Read<K, V> read) { super("KafkaIO.Read"); this.read = read; }

Should we use different names for easier debugging?

ihji · 2021-06-25T06:53:49Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java

+   * record for external transforms. TODO(BEAM-7345): use regular KafkaRecord class when Beam Schema
+   * inference supports generics.
+   */
+  static class ExternalKafkaRecord {


I think ExternalKafkaRecord is somewhat misleading. I would prefer to use something more intuitive (ex. RowCodedKafkaRecord, KafkaRowWithMetadata, etc.)

Changed to "ByteArrayKafkaRecord".

ihji · 2021-06-25T06:57:52Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java

+      // Using the transform name in the URN so that the corresponding transform can be easily
+      // identified.
+      public static final String URN_WITH_METADATA =
+          "beam:external:java:kafkaio:externalwithmetadata:v1";


Is it necessary to use external again in externalwithmetadata? external is already in the URN.

How about typedwithmetadata or rowwithmetadata?

Changed to "RowsWithMetadata".

chamikaramj

Thanks!

chamikaramj · 2021-06-26T02:38:26Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java

+   * record for external transforms. TODO(BEAM-7345): use regular KafkaRecord class when Beam Schema
+   * inference supports generics.
+   */
+  static class ExternalKafkaRecord {


Changed to "ByteArrayKafkaRecord".

chamikaramj · 2021-06-26T02:38:33Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java

+    private final Read<K, V> read;
+
+    ExternalWithMetadata(Read<K, V> read) {
+      super("KafkaIO.Read");


chamikaramj · 2021-06-26T02:39:06Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java

+      // Using the transform name in the URN so that the corresponding transform can be easily
+      // identified.
+      public static final String URN_WITH_METADATA =
+          "beam:external:java:kafkaio:externalwithmetadata:v1";


Changed to "RowsWithMetadata".

chamikaramj · 2021-06-26T03:51:33Z

Run Java PreCommit

chamikaramj · 2021-06-26T03:52:41Z

Run Portable_Python PreCommit

chamikaramj · 2021-06-26T16:28:54Z

Run Java PreCommit

chamikaramj · 2021-07-01T03:54:47Z

Run Python PreCommit

chamikaramj · 2021-07-01T04:32:02Z

Run Java PreCommit

chamikaramj · 2021-07-01T05:08:41Z

Run Python PreCommit

chamikaramj · 2021-07-01T05:08:47Z

Run PythonDocker PreCommit

chamikaramj · 2021-07-01T05:09:08Z

Run Portable_Python PreCommit

chamikaramj · 2021-07-01T07:58:50Z

Run Java PreCommit

chamikaramj · 2021-07-01T07:59:01Z

Run Python PreCommit

chamikaramj · 2021-07-01T15:54:57Z

Run Portable_Python PreCommit

chamikaramj · 2021-07-01T18:08:22Z

Run Portable_Python PreCommit failures seems to be unrelated (Execution failed for task ':sdks:go:installDependencies').

chamikaramj · 2021-07-01T18:09:49Z

Run Portable_Python PreCommit

Adds a Kafka external read transform for reading from Kafka with meta…

040b1eb

…data

chamikaramj force-pushed the py_kafka_with_metadata branch from d27fd99 to 040b1eb Compare June 21, 2021 14:38

chamikaramj changed the title ~~Adds a schema-friendly KafkaIO read transform that reads records with metadata~~ [BEAM-12076] Adds a Kafka external read transform for reading from Kafka with metadata Jun 21, 2021

chamikaramj changed the title ~~[BEAM-12076] Adds a Kafka external read transform for reading from Kafka with metadata~~ [BEAM-12076] Adds a Kafka external read transform for reading with metadata Jun 21, 2021

updates

0474b94

Fixes the example

ce2cabc

Updates URNs

b449e0c

ihji approved these changes Jun 25, 2021

View reviewed changes

Addresses reviewer comments

8277aa3

chamikaramj commented Jun 26, 2021

View reviewed changes

Changes the default to preserve backwards compatiblity

7906f96

chamikaramj merged commit d6088d3 into apache:master Jul 1, 2021

Conversation

chamikaramj commented Jun 17, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

ValidatesRunner compliance status (on master branch)

Examples testing status on various runners

Post-Commit SDK/Transform Integration Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

GitHub Actions Tests Status (on master branch)

Uh oh!

codecov bot commented Jun 17, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

chamikaramj commented Jun 21, 2021

Uh oh!

chamikaramj commented Jun 23, 2021

Uh oh!

chamikaramj commented Jun 24, 2021

Uh oh!

ihji 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!

chamikaramj 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!

chamikaramj commented Jun 26, 2021

Uh oh!

chamikaramj commented Jun 26, 2021

Uh oh!

chamikaramj commented Jun 26, 2021

Uh oh!

chamikaramj commented Jul 1, 2021

Uh oh!

chamikaramj commented Jul 1, 2021

Uh oh!

chamikaramj commented Jul 1, 2021

Uh oh!

chamikaramj commented Jul 1, 2021

Uh oh!

chamikaramj commented Jul 1, 2021

Uh oh!

chamikaramj commented Jul 1, 2021

Uh oh!

chamikaramj commented Jul 1, 2021

Uh oh!

chamikaramj commented Jul 1, 2021

Uh oh!

chamikaramj commented Jul 1, 2021

Uh oh!

chamikaramj commented Jul 1, 2021

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

chamikaramj commented Jun 17, 2021 •

edited

Loading

`ValidatesRunner` compliance status (on master branch)

codecov bot commented Jun 17, 2021 •

edited

Loading