[FLINK-12823][datastream] PartitionTransformation supports DataExchan… #8721

ifndef-SleePy · 2019-06-13T07:48:17Z

…geMode property

What is the purpose of the change

Since StreamTransformation would also support batch runner in 1.9, the result partition type of StreamTransformation should not be hard-coded with PIPELINED_BOUNDED
We need to provide a way for upper level API upon StreamTransformation to configure the result partition type of edge

Brief change log

Expose a property DataExchangeMode of PartitionTransformation as an internal API of StreamTransformation
Pass the DataExchangeMode to StreamEdge
StreamingJobGraphGenerator chooses the appropriate result partition type based on DataExchangeMode of StreamEdge

Verifying this change

Add an unit test of StreamingJobGraphGeneratorTest

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): no
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: no
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? no
If yes, how is the feature documented? not applicable

flinkbot · 2019-06-13T07:50:34Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❗ 3. Needs [attention] from.
- Needs attention by @aljoscha [PMC]
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

aljoscha · 2019-06-13T12:39:45Z

I think we should not reuse the existing DataExchangeMode because it is a runtime class in the flink-runtime package. At some point we want to decouple the StreamTransformations and the DataStream API from runtime so we should not introduce more dependencies. I would suggest to add a new enum next to PartitionTransformation (in the same package) for this purpose, it should probably also be declared @PublicEvolving.

ifndef-SleePy · 2019-06-14T07:23:36Z

Hi @aljoscha
Thanks for feedback. Good point! I have added a new ShuffleMode instead of reusing DataExchangeMode.

@flinkbot attention @aljoscha

aljoscha

This seems very good now. I had a comment about a test and about adding null checks for the new shuffle mode.

aljoscha · 2019-06-14T09:21:35Z

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java

@@ -428,7 +436,11 @@ private void addEdgeInternal(Integer upStreamVertexID,
 				}
 			}

-			StreamEdge edge = new StreamEdge(upstreamNode, downstreamNode, typeNumber, outputNames, partitioner, outputTag);
+			if (shuffleMode == null) {


Can it happen that shuffleMode is null? We should add a null check in PartitionTransform and mark the field @Nonnull.

Yes, it can be null here. Because there might be no PartitionTransformation if user does not specify explicitly. And then partitioner and shuffleMode would be null here.

Ok, makes sense

aljoscha · 2019-06-14T09:23:12Z

...va/src/main/java/org/apache/flink/streaming/api/transformations/PartitionTransformation.java

@@ -69,6 +88,15 @@ public PartitionTransformation(StreamTransformation<T> input, StreamPartitioner<
 		return partitioner;
 	}

+	/**


I don't like these redundant descriptions to much, this can be simplified to:

/** * Returns the {@link ShuffleMode} of this {@link PartitionTransformation}. */

aljoscha · 2019-06-14T09:23:39Z

...va/src/main/java/org/apache/flink/streaming/api/transformations/PartitionTransformation.java

@@ -79,6 +107,6 @@ public PartitionTransformation(StreamTransformation<T> input, StreamPartitioner<

 	@Override
 	public final void setChainingStrategy(ChainingStrategy strategy) {
-		throw new UnsupportedOperationException("Cannot set chaining strategy on Union Transformation.");
+		throw new UnsupportedOperationException("Cannot set chaining strategy on Partition Transformation.");


Nice catch! 😅

aljoscha · 2019-06-14T09:24:23Z

...-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java

@@ -133,8 +136,8 @@ public void testChainStartEndSetting() throws Exception {
 				public Integer map(Integer value) throws Exception {
 					return value;
 				}
-			})
-			.print();
+			}).setParallelism(2)


Is this change needed? It seems to be an unrelated test.

It's an unrelated change in a sense.
I found it when adding unit test. It's a potential unstable case since it depends on the StreamExecutionEnvironment.defaultLocalParallelism which must be not 1 in this case.
However I just found someone has fixed it yesterday. What a coincident! I can remove it from this PR.

ifndef-SleePy · 2019-06-14T09:57:53Z

Hi @aljoscha
I have rebased master and addressed comments.

aljoscha · 2019-06-14T10:59:38Z

Thanks! I think I can merge once Travis is green. 👌

ifndef-SleePy · 2019-06-17T02:17:55Z

Travis failed, there is an importing error in python module. I don't think it is caused by this PR. I will try to rebase master and trigger travis checking again.

…formation

ifndef-SleePy · 2019-06-17T05:31:17Z

Python module is fine but connectors module failed this time on Travis checking. It seems a travis relevant issue. I have filed a JIRA issue for this, https://issues.apache.org/jira/browse/FLINK-12866
Will trigger checking again through closing and reopening, hope it works.

aljoscha · 2019-06-17T11:43:08Z

I merged this. Thanks for working on this with me! 👍

rmetzger added review=description? component=API/DataStream labels Jun 13, 2019

aljoscha reviewed Jun 14, 2019

View reviewed changes

ifndef-SleePy force-pushed the FLINK-12823 branch from e76bd0e to 52e9230 Compare June 14, 2019 09:56

[FLINK-12823][datastream] Add ScheduleMode property of PartitionTrans…

3cfd7b7

…formation

ifndef-SleePy force-pushed the FLINK-12823 branch from 52e9230 to 3cfd7b7 Compare June 17, 2019 02:19

ifndef-SleePy closed this Jun 17, 2019

ifndef-SleePy reopened this Jun 17, 2019

aljoscha closed this Jun 17, 2019

ifndef-SleePy deleted the FLINK-12823 branch June 17, 2019 13:00

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-12823][datastream] PartitionTransformation supports DataExchan… #8721

[FLINK-12823][datastream] PartitionTransformation supports DataExchan… #8721

ifndef-SleePy commented Jun 13, 2019

flinkbot commented Jun 13, 2019 •

edited

aljoscha commented Jun 13, 2019

ifndef-SleePy commented Jun 14, 2019 •

edited

aljoscha left a comment

aljoscha Jun 14, 2019

ifndef-SleePy Jun 14, 2019 •

edited

aljoscha Jun 14, 2019

aljoscha Jun 14, 2019

ifndef-SleePy Jun 14, 2019

aljoscha Jun 14, 2019

aljoscha Jun 14, 2019

ifndef-SleePy Jun 14, 2019 •

edited

ifndef-SleePy commented Jun 14, 2019

aljoscha commented Jun 14, 2019

ifndef-SleePy commented Jun 17, 2019 •

edited

ifndef-SleePy commented Jun 17, 2019 •

edited

aljoscha commented Jun 17, 2019

[FLINK-12823][datastream] PartitionTransformation supports DataExchan… #8721

[FLINK-12823][datastream] PartitionTransformation supports DataExchan… #8721

Conversation

ifndef-SleePy commented Jun 13, 2019

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented Jun 13, 2019 • edited

Review Progress

aljoscha commented Jun 13, 2019

ifndef-SleePy commented Jun 14, 2019 • edited

aljoscha left a comment

Choose a reason for hiding this comment

aljoscha Jun 14, 2019

Choose a reason for hiding this comment

ifndef-SleePy Jun 14, 2019 • edited

Choose a reason for hiding this comment

aljoscha Jun 14, 2019

Choose a reason for hiding this comment

aljoscha Jun 14, 2019

Choose a reason for hiding this comment

ifndef-SleePy Jun 14, 2019

Choose a reason for hiding this comment

aljoscha Jun 14, 2019

Choose a reason for hiding this comment

aljoscha Jun 14, 2019

Choose a reason for hiding this comment

ifndef-SleePy Jun 14, 2019 • edited

Choose a reason for hiding this comment

ifndef-SleePy commented Jun 14, 2019

aljoscha commented Jun 14, 2019

ifndef-SleePy commented Jun 17, 2019 • edited

ifndef-SleePy commented Jun 17, 2019 • edited

aljoscha commented Jun 17, 2019

flinkbot commented Jun 13, 2019 •

edited

ifndef-SleePy commented Jun 14, 2019 •

edited

ifndef-SleePy Jun 14, 2019 •

edited

ifndef-SleePy Jun 14, 2019 •

edited

ifndef-SleePy commented Jun 17, 2019 •

edited

ifndef-SleePy commented Jun 17, 2019 •

edited