Initial prototype of using Beam transforms directly in a Flink pipeline. #30332

robertwb · 2024-02-15T20:22:26Z

Handling DataStream, windowing, and more complex types will come in a future PR.

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

Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
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.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

Handling DataStream, windowing, and more complex types will come in a future PR.

robertwb · 2024-02-15T20:29:55Z

R: @kennknowles

github-actions · 2024-02-15T20:31:54Z

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control

robertwb · 2024-02-15T21:00:46Z

(Looks like I'm going to have to rebase this on the org.apache.beam.runners.core.construction move...)

robertwb · 2024-02-15T21:15:16Z

(Looks like I'm going to have to rebase this on the org.apache.beam.runners.core.construction move...)

Done. PTAL.

robertwb · 2024-02-26T17:28:19Z

Ping.

kennknowles

My overall comment is just to tighten it up by removing suppressions and looking to eliminate implementation inheritance.

kennknowles · 2024-02-16T02:40:09Z

runners/flink/src/main/java/org/apache/beam/runners/flink/adapter/BeamAdapterUtils.java

+public class BeamAdapterUtils {
+  private BeamAdapterUtils() {}
+
+  @SuppressWarnings("nullness")


This much cyclomatic complexity and you want to suppress null checks? 😛

I tried to suppress them as locally as possible.

I've changed this to explicitly assert non-nullness here, but I'm not sure it's an improvement.

Local checks/assertions are better, because they don't mask errors that might be introduced as the code is edited. We should aim for no broken windows / zero suppressions. Certainly in new code. Any suppression should be linked to a bug with an explanation for why it cannot be addressed.

Using org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull or org.apache.beam.sdkutil.Preconditions.checkStateNotNull are both good options. They convert NPE at some random point in code to an IllegalArgumentException or IllegalStateException at the point of detection.

kennknowles · 2024-02-16T02:40:32Z

runners/flink/src/main/java/org/apache/beam/runners/flink/adapter/BeamAdapterUtils.java

+    }
+  }
+
+  static <T> TypeInformation<T> coderTotoTypeInformation(Coder<T> coder, PipelineOptions options) {


kennknowles · 2024-02-16T02:43:11Z

runners/flink/src/main/java/org/apache/beam/runners/flink/adapter/BeamFlinkAbstractAdapter.java

+import org.apache.flink.api.java.ExecutionEnvironment;
+
+public abstract class BeamFlinkAbstractAdapter<DataSetOrStream> {
+  protected final PipelineOptions pipelineOptions;


My take here is that this superclass is sensible standalone logic that is not in a mutually-recursive relationship with the subclass. So remove the subclassing and pass the { getTypeInformation; createContext } thing in to the logic here.

Kind of a stylistic point, but still, protected fields where the values are passed through the super constructor anti-pattern called attention to it.

Plus, then the stricter separation between Dataset / DataStream parameterization will be better, and this primary logic won't have it bleed in.

Unless there is a mutual recursion I missed... which also I would have missed because the control flow between subclass and superclass is hard to follow, hence good to eliminate.

I was able to extract it out. (This was an artifact of splitting the class up on the generic vs. type-specific bits.) I think this is better now, though it does give one method that has a ton of parameters...

I also pulled out FlinkInput/FlinkOutput, but we could make them inner classes again if you'd prefer.

kennknowles · 2024-02-16T02:46:24Z

runners/flink/src/main/java/org/apache/beam/runners/flink/adapter/BeamFlinkAbstractAdapter.java

+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.ExecutionEnvironment;
+
+public abstract class BeamFlinkAbstractAdapter<DataSetOrStream> {


T suffix would be traditional to indicate it is a type variable (since java doesn't have lexicographic distinction)

kennknowles · 2024-02-16T02:59:26Z

...s/flink/src/test/java/org/apache/beam/runners/flink/adapter/BeamFlinkDataSetAdapterTest.java

+
+    DataSet<String> input = env.fromCollection(ImmutableList.of("a", "b", "c"));
+    DataSet<String> result =
+        new BeamFlinkDataSetAdapter(PipelineOptionsFactory.create(), env)


I'd say we should try to find a way to polish this into an extremely minimal API as we go:

make the pipeline options optional and default them internal to the adapter

extract the env from the inputs so you don't have to pass it here

Ideally we could get very close to this signature:

DataSet<OutputT> applyBeamTransform(DataSet<InputT>, PTransform<InputT, OutputT>)

(which of course is begging to be upstreamed into DataSet API)

+1

PipelineOptions are now optional. I had initially thought of extracting the env from the input, but this is not possible transforms with no inputs. I actually don't think that's so bad for the Reads (it's similar to what one needs to do for Flink anyway) but this now precludes empty tuples. I think these are rare, we could add an overload taking the environment explicitly if need be.

I had considered going all the way and making these static methods, but I want to reserve the right to store any other state here (including possibly across invocations) and this'll allow for people to choose shorter variable names too.

kennknowles · 2024-02-16T03:01:50Z

runners/flink/src/main/java/org/apache/beam/runners/flink/adapter/BeamFlinkAbstractAdapter.java

+    translator.translate(translationContext, translator.prepareForTranslation(pipelineProto));
+  }
+
+  static class FlinkInput<T> extends PTransform<PBegin, PCollection<T>> {


I don't recall how the visibility of this resolves. Perhaps a minor thing but it would be nice for this to be package-private. In fact ideally everything is package private except just the call to applyBeamTransform(DataSet, PTransform)

These are package private. applyBeamTransform is the only thing I want to be public.

kennknowles · 2024-02-16T03:03:54Z

runners/flink/src/main/java/org/apache/beam/runners/flink/adapter/BeamFlinkAbstractAdapter.java

+import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.util.construction.Environments;
+import org.apache.beam.sdk.util.construction.NativeTransforms;


I had forgotten about this or missed it. That's fun...

Yeah. Means there's possibly issues linking in more than one runner though...

kennknowles · 2024-02-26T18:48:07Z

Otherwise LGTM as far as getting this going. Very cool!

robertwb

Thanks for your thoughtful review. PTAL.

robertwb · 2024-02-27T00:40:09Z

runners/flink/src/main/java/org/apache/beam/runners/flink/adapter/BeamAdapterUtils.java

+    }
+  }
+
+  static <T> TypeInformation<T> coderTotoTypeInformation(Coder<T> coder, PipelineOptions options) {


robertwb · 2024-02-27T00:46:12Z

runners/flink/src/main/java/org/apache/beam/runners/flink/adapter/BeamAdapterUtils.java

+public class BeamAdapterUtils {
+  private BeamAdapterUtils() {}
+
+  @SuppressWarnings("nullness")


I tried to suppress them as locally as possible.

I've changed this to explicitly assert non-nullness here, but I'm not sure it's an improvement.

robertwb · 2024-02-27T00:46:51Z

runners/flink/src/main/java/org/apache/beam/runners/flink/adapter/BeamFlinkAbstractAdapter.java

+import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.util.construction.Environments;
+import org.apache.beam.sdk.util.construction.NativeTransforms;


Yeah. Means there's possibly issues linking in more than one runner though...

robertwb · 2024-02-27T00:51:44Z

...s/flink/src/test/java/org/apache/beam/runners/flink/adapter/BeamFlinkDataSetAdapterTest.java

+
+    DataSet<String> input = env.fromCollection(ImmutableList.of("a", "b", "c"));
+    DataSet<String> result =
+        new BeamFlinkDataSetAdapter(PipelineOptionsFactory.create(), env)


+1

PipelineOptions are now optional. I had initially thought of extracting the env from the input, but this is not possible transforms with no inputs. I actually don't think that's so bad for the Reads (it's similar to what one needs to do for Flink anyway) but this now precludes empty tuples. I think these are rare, we could add an overload taking the environment explicitly if need be.

I had considered going all the way and making these static methods, but I want to reserve the right to store any other state here (including possibly across invocations) and this'll allow for people to choose shorter variable names too.

robertwb · 2024-02-27T00:52:17Z

runners/flink/src/main/java/org/apache/beam/runners/flink/adapter/BeamFlinkAbstractAdapter.java

+    translator.translate(translationContext, translator.prepareForTranslation(pipelineProto));
+  }
+
+  static class FlinkInput<T> extends PTransform<PBegin, PCollection<T>> {


These are package private. applyBeamTransform is the only thing I want to be public.

robertwb · 2024-02-27T00:53:24Z

runners/flink/src/main/java/org/apache/beam/runners/flink/adapter/BeamFlinkAbstractAdapter.java

+import org.apache.flink.api.java.ExecutionEnvironment;
+
+public abstract class BeamFlinkAbstractAdapter<DataSetOrStream> {
+  protected final PipelineOptions pipelineOptions;


I was able to extract it out. (This was an artifact of splitting the class up on the generic vs. type-specific bits.) I think this is better now, though it does give one method that has a ton of parameters...

I also pulled out FlinkInput/FlinkOutput, but we could make them inner classes again if you'd prefer.

robertwb · 2024-03-04T18:54:12Z

I've addressed all your comments so I'm going to go on your LGTM and merge this.

kennknowles · 2024-03-04T19:59:04Z

There are still lots of nullness checks suppressed that presumably should be fixed, not suppressed. I didn't see any systemic reason to disable checking.

robertwb · 2024-03-04T20:59:07Z

Most of these are because Map.get() uses nulls rather than exceptions to indicate keys aren't there (that tautologically should be). Cleaned up at #30488

github-actions bot added runners flink fn-execution labels Feb 15, 2024

Initial prototype of using Beam transforms directly in a Flink pipeline.

d4add30

Handling DataStream, windowing, and more complex types will come in a future PR.

robertwb force-pushed the flink-on-beam branch from dd133e2 to d4add30 Compare February 15, 2024 20:29

robertwb requested a review from kennknowles February 15, 2024 20:29

Remove length prefix workaround for now.

47c1d31

github-actions bot removed the fn-execution label Feb 15, 2024

fix runner construction rebase issues

9a72895

conform to type param naming rules

029964e

kennknowles requested changes Feb 26, 2024

View reviewed changes

robertwb added 2 commits February 26, 2024 16:53

First round of refactoring.

77a94c7

Move classes around now that we no longer have inheritance.

bce07bf

github-actions bot added the website label Feb 27, 2024

robertwb commented Feb 27, 2024

View reviewed changes

Remove irrelevent changes (not sure what tool is doing these.)

564d4d2

robertwb force-pushed the flink-on-beam branch from ef26f68 to 564d4d2 Compare February 27, 2024 01:28

Merge branch 'master' into flink-on-beam

177f017

github-actions bot removed the website label Feb 27, 2024

missing import

907758c

robertwb merged commit 9c3f209 into apache:master Mar 4, 2024
22 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Initial prototype of using Beam transforms directly in a Flink pipeline. #30332

Initial prototype of using Beam transforms directly in a Flink pipeline. #30332

robertwb commented Feb 15, 2024

robertwb commented Feb 15, 2024

github-actions bot commented Feb 15, 2024

robertwb commented Feb 15, 2024

robertwb commented Feb 15, 2024

robertwb commented Feb 26, 2024

kennknowles left a comment

kennknowles Feb 16, 2024

robertwb Feb 27, 2024

kennknowles Feb 27, 2024

kennknowles Feb 16, 2024

robertwb Feb 27, 2024

kennknowles Feb 16, 2024

robertwb Feb 27, 2024

kennknowles Feb 16, 2024

kennknowles Feb 16, 2024

robertwb Feb 27, 2024

kennknowles Feb 16, 2024

robertwb Feb 27, 2024

kennknowles Feb 16, 2024

robertwb Feb 27, 2024

kennknowles commented Feb 26, 2024

robertwb left a comment

robertwb Feb 27, 2024

robertwb Feb 27, 2024

robertwb Feb 27, 2024

robertwb Feb 27, 2024

robertwb Feb 27, 2024

robertwb Feb 27, 2024

robertwb commented Mar 4, 2024

kennknowles commented Mar 4, 2024

robertwb commented Mar 4, 2024

Initial prototype of using Beam transforms directly in a Flink pipeline. #30332

Initial prototype of using Beam transforms directly in a Flink pipeline. #30332

Conversation

robertwb commented Feb 15, 2024

GitHub Actions Tests Status (on master branch)

robertwb commented Feb 15, 2024

github-actions bot commented Feb 15, 2024

robertwb commented Feb 15, 2024

robertwb commented Feb 15, 2024

robertwb commented Feb 26, 2024

kennknowles left a 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

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

kennknowles commented Feb 26, 2024

robertwb left a 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

robertwb commented Mar 4, 2024

kennknowles commented Mar 4, 2024

robertwb commented Mar 4, 2024