[FLINK-19317] Make EventTime the default TimeCharacteristic #13509

aljoscha · 2020-09-29T13:23:37Z

What is the purpose of the change

This is part of http://s.apache.org/FLIP-134 (Batch execution for the DataStream API).

Brief change log

Please look at each commit message which describes in detail what is happening.

In commit no 1 I also change all usage of timeWindow() to use explicit processing-time or event-time windows. For this, I temporarily removed timeWindow() from the API and made sure everything compiles afterward. You can follow the process because I still left the removal and the revert of the removal in the commits.

commit no 1: make EventTime the new default: here I need to fix some tests which assumed the previous default behaviour
commit no 2: undo temporary removal of timeWindow()
commit no 3: remove unnecessary calls to setStreamTimeCharacteristic (java). I'm removing almost all calls because we're planning to deprecate/remove the method
commit no 4: remove unnecessary calls to setStreamTimeCharacteristic (scala)
commit no 5: update docs for default changes, here I remove almost all (all?) mentions of the time characteristic
commit no 6: deprecate setStreamTimeCharacteristic, this also adds a good deprecation message

Verifying this change

Existing tests cover the changes.

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): yes
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: no
The S3 file system connector: no

Documentation

I fix the javadocs, introduce a good deprecation text for setStreamTimeCharacteristic() and update the docs.

flinkbot · 2020-09-29T13:25:58Z

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.

Automated Checks

Last check on commit a0a68de (Fri Feb 19 07:32:33 UTC 2021)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 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.

Details

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

flinkbot · 2020-09-29T13:46:20Z

CI report:

a0a68de Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

dawidwys

Overall looks good!

Shall we remove also references to timeWindow from docs? There are quite a few of them.

Side note: have you tried removing the setTimeCharacteristics? Basically the same experiment as you did for timeWindow? I'd be curious, how easy it would, because we are deprecating the TimeCharacteristics.

docs/dev/event_time.md

...amples-streaming/src/main/java/org/apache/flink/streaming/examples/async/AsyncIOExample.java

dawidwys · 2020-09-29T17:29:48Z

...rc/main/scala/org/apache/flink/streaming/scala/examples/ml/IncrementalLearningSkeleton.scala

    // build new model on every second of new data
-    val trainingData: DataStream[Int] = env.addSource(new FiniteTrainingDataSource)
-    val newData: DataStream[Int] = env.addSource(new FiniteNewDataSource)
+    val trainingData: DataStream[Integer] = env.addSource(new FiniteTrainingDataSource)


Why did we have to change to java classes here?

The reason is that the explicit window assigners are usually a WindowAssigner<Object, TimeWindow>. This doesn't work for Scala Int, Long, etc. because they don't extend from Object but from AnyVal, so the Scala compiler will not allow it. It's not pretty, but I think the example is pretty artificial and in any real-world use case we would never have a DataStream of primitives.

I see. Thanks for the explanation.

dawidwys · 2020-09-29T17:33:37Z

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/TimeCharacteristic.java

+ * The time characteristic defines how the system determines time for time-dependent order and
+ * operations that depend on time (such as time windows).
+ *
+ * @deprecated In Flink 1.12 the default stream time characteristic has been changed to {@link


nit: I am not entirely sure, but I think 1.12 is the first release with python DataStream API. If that's the case I think we don't need to mention the change of the default value here.

It doesn't harm though, therefore I am fine with either.

Why do you mention Python here?

Sorry, my bad. Before this class there were python files and I somehow confused this with the python equivalent.

BTW, do you think we should deprecate the setStreamTimeCharacteristic in python as well (or even remove it, as it was added in 1.12?)

dawidwys · 2020-09-30T06:33:28Z

BTW, I forgot to mention I really liked commit messages!

kl0u · 2020-09-30T07:01:35Z

@aljoscha I am looking at the PR now, but I was wondering if we can deprecate the TimeCharacteristic given that we still use it in the StreamConfig to determine stuff on the cluster side, e.g. the SourceContext. We may be able to also remove it from there in the future and use, for example, the watermarkInterval to determine the source context.

kl0u

I had some minor comments on the PR. I cannot spot any serious problem so feel free to integrate as many of my comments as you think and merge after Azure gives green.

kl0u · 2020-09-30T07:28:07Z

...ava/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java

+	 *        TimeCharacteristic#EventTime}, thus you don't need to call this method for enabling
+	 * 		event-time support anymore. Explicitly using processing-time windows and timers works in
+	 * 		event-time mode. If you need to disable watermarks, please use {@link
+	 *        ExecutionConfig#setAutoWatermarkInterval(long)}. If you are using {@link


I think it is useful to also mention explicitly in the doc the value to set the watermark interval in order to use ProcessingTime. Sth like: "If you need to disable watermarks, please set the {@code watermarkInterval} to 0 using the {@link ExecutionConfig#setAutoWatermarkInterval(long)}."

Good point! I would update the Javadoc of setAutoWatermarkInterval() to mention that setting it to 0 will disable it. I'm reluctant to add it here because the message is already very long. WDYT?

I forgot to say that the

else { getConfig().setAutoWatermarkInterval(200); }

is not strictly needed anymore, as 200 is the new default value.

kl0u · 2020-09-30T07:28:59Z

flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java


 	private boolean forceAvro = false;
-	private long autoWatermarkInterval = 0;
+	private long autoWatermarkInterval = 200;


nit: Make the 200 a static var with a descriptive name instead of a "magic" number.

I can add a static final DEFAULT_WATERMARK_INTERVAL but then it would basically be

private static final long DEFAULT_WATERMARK_INTERVAL = 200; private long autoWatermarkInterval = DEFAULT_WATERMARK_INTERVAL;

except that we might but the default at the top. If you think it would help I can do it.

I do not think it contributes to anything for now. It is just that if in the future someone, for whatever reason, wants to do a check like: if (interval == DEFAULT) ..., or he/she wants to reset the interval to the default, the literal 200 would be a bit obscure instead of a DEFAULT_WM_INTERVAL. But again, it does not add any value for now.

kl0u · 2020-09-30T07:30:30Z

...treaming/src/main/java/org/apache/flink/streaming/examples/sideoutput/SideOutputExample.java

+	 * In a real use case you should use proper timestamps and an appropriate {@link
+	 * WatermarkStrategy}.
+	 */
+	private static class IngestionTimeWatermarkStrategy<T> implements WatermarkStrategy<T> {


This WatermarkStrategy seems to be used extensively in the tests. Couldn't we extract it somewhere and make it reusable?

I had this in the commit message:

I removed calls to set IngestionTime and replaced them by an explicit IngestionTimeWatermarkStrategy. I duplicated the same IngestionTimeWatermarkStrategy in all the examples/tests because I explicitly didn't want to add an IngestionTimeWatermarkStrategy in one of the core packages so that it is not discoverable because I think we shouldn't encourage users to use ingestion time.

what do you think?

I was thinking more of a testing module, like the MockSource, but I think you are right to not expose it even there.

aljoscha · 2020-09-30T08:32:00Z

Thanks for the reviews!

@dawidwys I didn't try removing setStreamTimeCharacteristic but I manually checked the references. There are a couple of clusters where it's still used:

tests that verify explicit behaviour based on the characteristic, for example in the Table Planner tests or the time window translation tests
CEP tests, but there's a follow-up issue for addressing that: https://issues.apache.org/jira/browse/FLINK-19326
time interval join (tests), unfortunately they also magically change behaviour based on the characteristic: https://issues.apache.org/jira/browse/FLINK-19479

@kl0u I'm also torn on deprecating TimeCharacteristic because we still use it internally. I did deprecate it because that's the message we want to send. What do you think?

kl0u · 2020-09-30T08:49:42Z

@aljoscha My only point was that I think that the message to the user to not set the time characteristic is sent by deprecating the setTimeCharacteristic(). My concern is that the next dev who uses the TimeCharacteristic will see the deprecation and he/she may be puzzled, as the deprecation in that case does not refer to the class but rather its exposure to the public API.

This is part of the FLIP-134 (Batch execution for the DataStream API) work. Event time is the only sensible time characteristic for batch processing. We therefore change the default value of the TimeCharacteristic from ProcessingTime to EventTime. This means the DataStream API programs that were using event time before now just work without manually changing this setting. Processing-time programs will also still work, because using processing-time timers is not dependent on the TimeCharacteristic. DataStream programs that don't set a TimestampAssigner or WatermarkStrategy will also still work if they don't use operations that don't rely on (event-time) timestamps. This is true for both BATCH and STREAMING execution mode. With this change, users don't need to call setStreamTimeCharacteristic(EventTime) anymore. We will make sure they learn of this by deprecating the method in a follow-up commit. The only real user-visible change of this is that programs that used the KeyedStream.timeWindow()/DataStream.timeWindow() operation, which is dependent on the TimeCharacteristic will now use event time by default. We don't think this operation is useful because the behaviour can be surprising. We recommend users always use an explicit processing-time window or event-time window. We also change the default watermark interval from 0 (disabled) to 200 to match the previous behaviour of calling setStreamTimeCharacteristic(EventTime).

… (java) I'm just removing calls the set EventTime because that's the new default now. I'm also removing most calls to set ProcessingTime because it's not needed for making processing-time timers/windows work. I only left it for some tests that check specific failure behavior. I removed calls to set IngestionTime and replaced them by an explicit IngestionTimeWatermarkStrategy. I duplicated the same IngestionTimeWatermarkStrategy in all the examples/tests because I explicitly didn't want to add an IngestionTimeWatermarkStrategy in one of the core packages so that it is not discoverable because I think we shouldn't encourage users to use ingestion time.

… (scala) I'm just removing calls the set EventTime because that's the new default now. I'm also removing most calls to set ProcessingTime because it's not needed for making processing-time timers/windows work. I only left it for some tests that check specific failure behavior. I removed calls to set IngestionTime and replaced them by an explicit IngestionTimeWatermarkStrategy. I duplicated the same IngestionTimeWatermarkStrategy in all the examples/tests because I explicitly didn't want to add an IngestionTimeWatermarkStrategy in one of the core packages so that it is not discoverable because I think we shouldn't encourage users to use ingestion time.

I remove most usages where EventTime is set and remove copy that talks about the default. In general, we should discourage using the stream time characteristic as much as possible.

…eristic After FLINK-19317 and FLINK-19318 we don't need this setting anymore. Using (explicit) processing-time windows and processing-time timers work fine in a program that has EventTime set as a time characteristic and once we deprecate timeWindow() there are not other operations that change behaviour depending on the time characteristic so there's no need to ever change from the new default of event-time. Similarly, the IngestionTime setting can be achieved in the future by providing an ingestion-time WatermarkStrategy.

aljoscha · 2020-09-30T11:38:27Z

Regarding TimeCharacteristic: I think that is the right message, I don’t want devs to use it because we should also remove it eventually.

aljoscha · 2020-10-01T15:15:08Z

Thanks for the reviews! I merged.

aljoscha requested review from dawidwys and kl0u September 29, 2020 13:23

rmetzger added review=description? component=API/DataStream labels Sep 29, 2020

kl0u self-assigned this Sep 29, 2020

dawidwys approved these changes Sep 29, 2020

View reviewed changes

kl0u approved these changes Sep 30, 2020

View reviewed changes

aljoscha added 5 commits September 30, 2020 12:23

[FLINK-19317] Update docs for new stream time characteristic default

3a4392d

I remove most usages where EventTime is set and remove copy that talks about the default. In general, we should discourage using the stream time characteristic as much as possible.

aljoscha force-pushed the flink-19317-event-time-default-characteristic branch from bc7bf58 to a0a68de Compare September 30, 2020 10:24

aljoscha closed this Oct 1, 2020

aljoscha deleted the flink-19317-event-time-default-characteristic branch October 1, 2020 15:14

[FLINK-19317] Make EventTime the default TimeCharacteristic #13509

[FLINK-19317] Make EventTime the default TimeCharacteristic #13509

Uh oh!

Conversation

aljoscha commented Sep 29, 2020

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

Uh oh!

flinkbot commented Sep 29, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Automated Checks

Review Progress

Uh oh!

flinkbot commented Sep 29, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

dawidwys left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dawidwys Sep 30, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

dawidwys commented Sep 30, 2020

Uh oh!

kl0u commented Sep 30, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

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

kl0u Sep 30, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

aljoscha commented Sep 30, 2020

Uh oh!

kl0u commented Sep 30, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

aljoscha commented Sep 30, 2020

Uh oh!

aljoscha commented Oct 1, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

flinkbot commented Sep 29, 2020 •

edited

Loading

flinkbot commented Sep 29, 2020 •

edited

Loading

dawidwys Sep 30, 2020 •

edited

Loading

kl0u commented Sep 30, 2020 •

edited

Loading

kl0u Sep 30, 2020 •

edited

Loading

kl0u commented Sep 30, 2020 •

edited

Loading