You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by uc...@apache.org on 2017/02/26 17:22:22 UTC
flink git commit: [FLINK-5896][docs] improve readability of event
time docs
Repository: flink
Updated Branches:
refs/heads/master 0a97cd29a -> c57f07b57
[FLINK-5896][docs] improve readability of event time docs
This closes #3403.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/c57f07b5
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/c57f07b5
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/c57f07b5
Branch: refs/heads/master
Commit: c57f07b575a6c9c36968f1c3ac574a4ce2eb28dd
Parents: 0a97cd2
Author: David Anderson <da...@singularity.local>
Authored: Thu Feb 23 16:23:50 2017 +0100
Committer: Ufuk Celebi <uc...@apache.org>
Committed: Sun Feb 26 18:22:09 2017 +0100
----------------------------------------------------------------------
docs/dev/event_time.md | 87 ++++++++++++++--------------
docs/dev/event_timestamp_extractors.md | 15 ++---
docs/dev/event_timestamps_watermarks.md | 79 ++++++++++++-------------
3 files changed, 90 insertions(+), 91 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/c57f07b5/docs/dev/event_time.md
----------------------------------------------------------------------
diff --git a/docs/dev/event_time.md b/docs/dev/event_time.md
index 83151da..0d3acaf 100644
--- a/docs/dev/event_time.md
+++ b/docs/dev/event_time.md
@@ -54,37 +54,37 @@ Flink supports different notions of *time* in streaming programs.
Event time gives correct results even on out-of-order events, late events, or on replays
of data from backups or persistent logs. In event time, the progress of time depends on the data,
not on any wall clocks. Event time programs must specify how to generate *Event Time Watermarks*,
- which is the mechanism that signals time progress in event time. The mechanism is
+ which is the mechanism that signals progress in event time. The mechanism is
described below.
Event time processing often incurs a certain latency, due to its nature of waiting a certain time for
late events and out-of-order events. Because of that, event time programs are often combined with
*processing time* operations.
-- **Ingestion time:** Ingestion time is the time that events enter Flink. At the source operator, each
+- **Ingestion time:** Ingestion time is the time that events enter Flink. At the source operator each
record gets the source's current time as a timestamp, and time-based operations (like time windows)
refer to that timestamp.
- *Ingestion Time* sits conceptually in between *Event Time* and *Processing Time*. Compared to
- *Processing Time*, it is slightly more expensive, but gives more predictable results: because
- *Ingestion Time* uses stable timestamps (assigned once at the source), different window operations
- over the records will refer to the same timestamp, whereas in *Processing Time* each window operator
+ *Ingestion time* sits conceptually in between *event time* and *processing time*. Compared to
+ *processing time*, it is slightly more expensive, but gives more predictable results. Because
+ *ingestion time* uses stable timestamps (assigned once at the source), different window operations
+ over the records will refer to the same timestamp, whereas in *processing time* each window operator
may assign the record to a different window (based on the local system clock and any transport delay).
- Compered to *Event Time*, *Ingestion Time* programs cannot handle any out-of-order events or late data,
- but the programs don't have to specify how to generate *Watermarks*.
+ Compared to *event time*, *ingestion time* programs cannot handle any out-of-order events or late data,
+ but the programs don't have to specify how to generate *watermarks*.
- Internally, *Ingestion Time* is treated much like event time, with automatic timestamp assignment and
- automatic Watermark generation.
+ Internally, *ingestion time* is treated much like *event time*, but with automatic timestamp assignment and
+ automatic watermark generation.
<img src="{{ site.baseurl }}/fig/times_clocks.svg" class="center" width="80%" />
### Setting a Time Characteristic
-The first part of a Flink DataStream program is usually to set the base *time characteristic*. That setting
-defines how data stream sources behave (for example whether to assign timestamps), and what notion of
-time the window operations like `KeyedStream.timeWindow(Time.seconds(30))` refer to.
+The first part of a Flink DataStream program usually sets the base *time characteristic*. That setting
+defines how data stream sources behave (for example, whether they will assign timestamps), and what notion of
+time should be used by window operations like `KeyedStream.timeWindow(Time.seconds(30))`.
The following example shows a Flink program that aggregates events in hourly time windows. The behavior of the
windows adapts with the time characteristic.
@@ -131,64 +131,64 @@ stream
</div>
-Note that in order to run this example in *Event Time*, the program needs to use either sources
-that directly define event time for the data and emits Watermarks themselves, or
+Note that in order to run this example in *event time*, the program needs to either use sources
+that directly define event time for the data and emit watermarks themselves, or the program must
inject a *Timestamp Assigner & Watermark Generator* after the sources. Those functions describe how to access
-the event timestamps, and what timely out-of-orderness the event stream exhibits.
+the event timestamps, and what degree of out-of-orderness the event stream exhibits.
-The section below describes the general mechanism behind *Timestamps* and *Watermarks*. For a guide on how
+The section below describes the general mechanism behind *timestamps* and *watermarks*. For a guide on how
to use timestamp assignment and watermark generation in the Flink DataStream API, please refer to
-[Generating Timestamps / Watermarks]({{ site.baseurl }}/dev/event_timestamps_watermarks.html)
+[Generating Timestamps / Watermarks]({{ site.baseurl }}/dev/event_timestamps_watermarks.html).
# Event Time and Watermarks
-*Note: Flink implements many techniques from the Dataflow Model. For a good introduction to Event Time and Watermarks, have also a look at the below articles.*
+*Note: Flink implements many techniques from the Dataflow Model. For a good introduction to event time and watermarks, have a look at the articles below.*
- [Streaming 101](https://www.oreilly.com/ideas/the-world-beyond-batch-streaming-101) by Tyler Akidau
- - The [Dataflow Model paper](https://static.googleusercontent.com/media/research.google.com/en//pubs/archive/43864.pdf)
+ - The [Dataflow Model paper](https://static.googleusercontent.com/media/research.google.com/en/pubs/archive/43864.pdf)
A stream processor that supports *event time* needs a way to measure the progress of event time.
-For example, a window operator that builds hourly windows needs to be notified when event time has reached the
-next full hour, such that the operator can close the next window.
+For example, a window operator that builds hourly windows needs to be notified when event time has passed beyond the
+end of an hour, so that the operator can close the window in progress.
-*Event Time* can progress independently of *Processing Time* (measured by wall clocks).
-For example, in one program, the current *event time* of an operator can trail slightly behind the processing time
-(accounting for a delay in receiving the latest elements) and both proceed at the same speed. In another streaming
-program, which reads fast-forward through some data already buffered in a Kafka topic (or another message queue), event time
-can progress by weeks in seconds.
+*Event time* can progress independently of *processing time* (measured by wall clocks).
+For example, in one program the current *event time* of an operator may trail slightly behind the *processing time*
+(accounting for a delay in receiving the events), while both proceed at the same speed.
+On the other hand, another streaming program might progress through weeks of event time with only a few seconds of processing,
+by fast-forwarding through some historic data already buffered in a Kafka topic (or another message queue).
------
-The mechanism in Flink to measure progress in event time is **Watermarks**.
+The mechanism in Flink to measure progress in event time is **watermarks**.
Watermarks flow as part of the data stream and carry a timestamp *t*. A *Watermark(t)* declares that event time has reached time
*t* in that stream, meaning that there should be no more elements from the stream with a timestamp *t' <= t* (i.e. events with timestamps
older or equal to the watermark).
-The figure below shows a stream of events with (logical) timestamps, and watermarks flowing inline. The events are in order
-(with respect to their timestamp), meaning that watermarks are simply periodic markers in the stream with an in-order timestamp.
+The figure below shows a stream of events with (logical) timestamps, and watermarks flowing inline. In this example the events are in order
+(with respect to their timestamps), meaning that the watermarks are simply periodic markers in the stream.
<img src="{{ site.baseurl }}/fig/stream_watermark_in_order.svg" alt="A data stream with events (in order) and watermarks" class="center" width="65%" />
-Watermarks are crucial for *out-of-order* streams, as shown in the figure below, where, events do not occur ordered by their timestamp.
-Watermarks establish points in the stream where all events up to a certain timestamp have occurred. Once these watermarks reach an
-operator, the operator can advance its internal *event time clock* to the value of the watermark.
+Watermarks are crucial for *out-of-order* streams, as illustrated below, where the events are not ordered by their timestamps.
+In general a watermark is a declaration that by that point in the stream, all events up to a certain timestamp should have arrived.
+Once a watermark reaches an operator, the operator can advance its internal *event time clock* to the value of the watermark.
<img src="{{ site.baseurl }}/fig/stream_watermark_out_of_order.svg" alt="A data stream with events (out of order) and watermarks" class="center" width="65%" />
## Watermarks in Parallel Streams
-Watermarks are generated at source functions, or directly after source functions. Each parallel subtask of a source function usually
+Watermarks are generated at, or directly after, source functions. Each parallel subtask of a source function usually
generates its watermarks independently. These watermarks define the event time at that particular parallel source.
As the watermarks flow through the streaming program, they advance the event time at the operators where they arrive. Whenever an
operator advances its event time, it generates a new watermark downstream for its successor operators.
-Operators that consume multiple input streams (e.g., after a *keyBy(...)* or *partition(...)* function, or a union) track the event time
-on each of their input streams. The operator's current event time is the minimum of the input streams' event time. As the input streams
-update their event time, so does the operator.
+Some operators consume multiple input streams; a union, for example, or operators following a *keyBy(...)* or *partition(...)* function.
+Such an operator's current event time is the minimum of its input streams' event times. As its input streams
+update their event times, so does the operator.
The figure below shows an example of events and watermarks flowing through parallel streams, and operators tracking event time.
@@ -197,15 +197,16 @@ The figure below shows an example of events and watermarks flowing through paral
## Late Elements
-It is possible that certain elements violate the watermark condition, meaning that even after the *Watermark(t)* has occurred,
+It is possible that certain elements will violate the watermark condition, meaning that even after the *Watermark(t)* has occurred,
more elements with timestamp *t' <= t* will occur. In fact, in many real world setups, certain elements can be arbitrarily
-delayed, making it impossible to define a time when all elements of a certain event timestamp have occurred.
-Further more, even if the lateness can be bounded, delaying the watermarks by too much is often not desirable, because it delays
-the evaluation of the event time windows by too much.
+delayed, making it impossible to specify a time by which all elements of a certain event timestamp will have occurred.
+Furthermore, even if the lateness can be bounded, delaying the watermarks by too much is often not desirable, because it
+causes too much delay in the evaluation of the event time windows.
-Due to that, some streaming programs will explicitly expect a number of *late* elements. Late elements are elements that
+For this reason, streaming programs may explicitly expect some *late* elements. Late elements are elements that
arrive after the system's event time clock (as signaled by the watermarks) has already passed the time of the late element's
-timestamp.
+timestamp. See [Allowed Lateness]({{ site.baseurl }}/dev/windows.html#allowed-lateness) for more information on how to work
+with late elements in event time windows.
## Debugging Watermarks
http://git-wip-us.apache.org/repos/asf/flink/blob/c57f07b5/docs/dev/event_timestamp_extractors.md
----------------------------------------------------------------------
diff --git a/docs/dev/event_timestamp_extractors.md b/docs/dev/event_timestamp_extractors.md
index e83f540..34a27ff 100644
--- a/docs/dev/event_timestamp_extractors.md
+++ b/docs/dev/event_timestamp_extractors.md
@@ -28,14 +28,14 @@ under the License.
As described in [timestamps and watermark handling]({{ site.baseurl }}/dev/event_timestamps_watermarks.html),
Flink provides abstractions that allow the programmer to assign their own timestamps and emit their own watermarks. More specifically,
one can do so by implementing one of the `AssignerWithPeriodicWatermarks` and `AssignerWithPunctuatedWatermarks` interfaces, depending
-on their use-case. In a nutshell, the first will emit watermarks periodically, while the second does so based on some property of
+on the use case. In a nutshell, the first will emit watermarks periodically, while the second does so based on some property of
the incoming records, e.g. whenever a special element is encountered in the stream.
In order to further ease the programming effort for such tasks, Flink comes with some pre-implemented timestamp assigners.
This section provides a list of them. Apart from their out-of-the-box functionality, their implementation can serve as an example
-for custom assigner implementations.
+for custom implementations.
-#### **Assigner with Ascending Timestamps**
+### **Assigners with ascending timestamps**
The simplest special case for *periodic* watermark generation is the case where timestamps seen by a given source task
occur in ascending order. In that case, the current timestamp can always act as a watermark, because no earlier timestamps will
@@ -43,7 +43,7 @@ arrive.
Note that it is only necessary that timestamps are ascending *per parallel data source task*. For example, if
in a specific setup one Kafka partition is read by one parallel data source instance, then it is only necessary that
-timestamps are ascending within each Kafka partition. Flink's Watermark merging mechanism will generate correct
+timestamps are ascending within each Kafka partition. Flink's watermark merging mechanism will generate correct
watermarks whenever parallel streams are shuffled, unioned, connected, or merged.
<div class="codetabs" markdown="1">
@@ -70,7 +70,7 @@ val withTimestampsAndWatermarks = stream.assignAscendingTimestamps( _.getCreatio
</div>
</div>
-#### **Assigner which allows a fixed amount of record lateness**
+### **Assigners allowing a fixed amount of lateness**
Another example of periodic watermark generation is when the watermark lags behind the maximum (event-time) timestamp
seen in the stream by a fixed amount of time. This case covers scenarios where the maximum lateness that can be encountered in a
@@ -78,8 +78,9 @@ stream is known in advance, e.g. when creating a custom source containing elemen
time for testing. For these cases, Flink provides the `BoundedOutOfOrdernessTimestampExtractor` which takes as an argument
the `maxOutOfOrderness`, i.e. the maximum amount of time an element is allowed to be late before being ignored when computing the
final result for the given window. Lateness corresponds to the result of `t - t_w`, where `t` is the (event-time) timestamp of an
-element, and `t_w` that of the previous watermark. If `lateness > 0` then the element is considered late and is ignored when computing
-the result of the job for its corresponding window.
+element, and `t_w` that of the previous watermark. If `lateness > 0` then the element is considered late and is, by default, ignored when computing
+the result of the job for its corresponding window. See the documentation about [allowed lateness]({{ site.baseurl }}/dev/windows.html#allowed-lateness)
+for more information about working with late elements.
<div class="codetabs" markdown="1">
<div data-lang="java" markdown="1">
http://git-wip-us.apache.org/repos/asf/flink/blob/c57f07b5/docs/dev/event_timestamps_watermarks.md
----------------------------------------------------------------------
diff --git a/docs/dev/event_timestamps_watermarks.md b/docs/dev/event_timestamps_watermarks.md
index a16aa55..0b8daf8 100644
--- a/docs/dev/event_timestamps_watermarks.md
+++ b/docs/dev/event_timestamps_watermarks.md
@@ -26,10 +26,10 @@ under the License.
{:toc}
-This section is relevant for program running on **Event Time**. For an introduction to *Event Time*,
-*Processing Time*, and *Ingestion Time*, please refer to the [event time introduction]({{ site.baseurl }}/dev/event_time.html)
+This section is relevant for programs running on **event time**. For an introduction to *event time*,
+*processing time*, and *ingestion time*, please refer to the [introduction to event time]({{ site.baseurl }}/dev/event_time.html).
-To work with *Event Time*, streaming programs need to set the *time characteristic* accordingly.
+To work with *event time*, streaming programs need to set the *time characteristic* accordingly.
<div class="codetabs" markdown="1">
<div data-lang="java" markdown="1">
@@ -48,14 +48,14 @@ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
## Assigning Timestamps
-In order to work with *Event Time*, Flink needs to know the events' *timestamps*, meaning each element in the
-stream needs to get its event timestamp *assigned*. That happens usually by accessing/extracting the
+In order to work with *event time*, Flink needs to know the events' *timestamps*, meaning each element in the
+stream needs to have its event timestamp *assigned*. This is usually done by accessing/extracting the
timestamp from some field in the element.
Timestamp assignment goes hand-in-hand with generating watermarks, which tell the system about
-the progress in event time.
+progress in event time.
-There are two ways to assign timestamps and generate Watermarks:
+There are two ways to assign timestamps and generate watermarks:
1. Directly in the data stream source
2. Via a timestamp assigner / watermark generator: in Flink timestamp assigners also define the watermarks to be emitted
@@ -65,14 +65,14 @@ millliseconds since the Java epoch of 1970-01-01T00:00:00Z.
### Source Functions with Timestamps and Watermarks
-Stream sources can also directly assign timestamps to the elements they produce and emit Watermarks. In that case,
-no Timestamp Assigner is needed.
+Stream sources can also directly assign timestamps to the elements they produce, and they can also emit watermarks.
+When this is done, no timestamp assigner is needed.
+Note that if a timestamp assigner is used, any timestamps and watermarks provided by the source will be overwritten.
To assign a timestamp to an element in the source directly, the source must use the `collectWithTimestamp(...)`
-method on the `SourceContext`. To generate Watermarks, the source must call the `emitWatermark(Watermark)` function.
+method on the `SourceContext`. To generate watermarks, the source must call the `emitWatermark(Watermark)` function.
-Below is a simple example of a source *(non-checkpointed)* that assigns timestamps and generates Watermarks
-depending on special events:
+Below is a simple example of a *(non-checkpointed)* source that assigns timestamps and generates watermarks:
<div class="codetabs" markdown="1">
<div data-lang="java" markdown="1">
@@ -106,17 +106,14 @@ override def run(ctx: SourceContext[MyType]): Unit = {
</div>
</div>
-*Note:* If the streaming program uses a TimestampAssigner on a stream where elements have a timestamp already,
-those timestamps will be overwritten by the TimestampAssigner. Similarly, Watermarks will be overwritten as well.
-
### Timestamp Assigners / Watermark Generators
-Timestamp Assigners take a stream and produce a new stream with timestamped elements and watermarks. If the
+Timestamp assigners take a stream and produce a new stream with timestamped elements and watermarks. If the
original stream had timestamps and/or watermarks already, the timestamp assigner overwrites them.
-The timestamp assigners usually are specified immediately after the data source but it is not strictly required to do so.
-A common pattern is, for example, to parse (*MapFunction*) and filter (*FilterFunction*) before the timestamp assigner.
+Timestamp assigners are usually specified immediately after the data source, but it is not strictly required to do so.
+A common pattern, for example, is to parse (*MapFunction*) and filter (*FilterFunction*) before the timestamp assigner.
In any case, the timestamp assigner needs to be specified before the first operation on event time
(such as the first window operation). As a special case, when using Kafka as the source of a streaming job,
Flink allows the specification of a timestamp assigner / watermark emitter inside
@@ -175,13 +172,13 @@ withTimestampsAndWatermarks
#### **With Periodic Watermarks**
-The `AssignerWithPeriodicWatermarks` assigns timestamps and generates watermarks periodically (possibly depending
+`AssignerWithPeriodicWatermarks` assigns timestamps and generates watermarks periodically (possibly depending
on the stream elements, or purely based on processing time).
The interval (every *n* milliseconds) in which the watermark will be generated is defined via
-`ExecutionConfig.setAutoWatermarkInterval(...)`. Each time, the assigner's `getCurrentWatermark()` method will be
-called, and a new Watermark will be emitted, if the returned Watermark is non-null and larger than the previous
-Watermark.
+`ExecutionConfig.setAutoWatermarkInterval(...)`. The assigner's `getCurrentWatermark()` method will be
+called each time, and a new watermark will be emitted if the returned watermark is non-null and larger than the previous
+watermark.
Two simple examples of timestamp assigners with periodic watermark generation are below.
@@ -189,9 +186,9 @@ Two simple examples of timestamp assigners with periodic watermark generation ar
<div data-lang="java" markdown="1">
{% highlight java %}
/**
- * This generator generates watermarks assuming that elements come out of order to a certain degree only.
- * The latest elements for a certain timestamp t will arrive at most n milliseconds after the earliest
- * elements for timestamp t.
+ * This generator generates watermarks assuming that elements arrive out of order,
+ * but only to a certain degree. The latest elements for a certain timestamp t will arrive
+ * at most n milliseconds after the earliest elements for timestamp t.
*/
public class BoundedOutOfOrdernessGenerator extends AssignerWithPeriodicWatermarks<MyEvent> {
@@ -214,8 +211,8 @@ public class BoundedOutOfOrdernessGenerator extends AssignerWithPeriodicWatermar
}
/**
- * This generator generates watermarks that are lagging behind processing time by a certain amount.
- * It assumes that elements arrive in Flink after at most a certain time.
+ * This generator generates watermarks that are lagging behind processing time by a fixed amount.
+ * It assumes that elements arrive in Flink after a bounded delay.
*/
public class TimeLagWatermarkGenerator extends AssignerWithPeriodicWatermarks<MyEvent> {
@@ -237,9 +234,9 @@ public class TimeLagWatermarkGenerator extends AssignerWithPeriodicWatermarks<My
<div data-lang="scala" markdown="1">
{% highlight scala %}
/**
- * This generator generates watermarks assuming that elements come out of order to a certain degree only.
- * The latest elements for a certain timestamp t will arrive at most n milliseconds after the earliest
- * elements for timestamp t.
+ * This generator generates watermarks assuming that elements arrive out of order,
+ * but only to a certain degree. The latest elements for a certain timestamp t will arrive
+ * at most n milliseconds after the earliest elements for timestamp t.
*/
class BoundedOutOfOrdernessGenerator extends AssignerWithPeriodicWatermarks[MyEvent] {
@@ -260,8 +257,8 @@ class BoundedOutOfOrdernessGenerator extends AssignerWithPeriodicWatermarks[MyEv
}
/**
- * This generator generates watermarks that are lagging behind processing time by a certain amount.
- * It assumes that elements arrive in Flink after at most a certain time.
+ * This generator generates watermarks that are lagging behind processing time by a fixed amount.
+ * It assumes that elements arrive in Flink after a bounded delay.
*/
class TimeLagWatermarkGenerator extends AssignerWithPeriodicWatermarks[MyEvent] {
@@ -282,15 +279,15 @@ class TimeLagWatermarkGenerator extends AssignerWithPeriodicWatermarks[MyEvent]
#### **With Punctuated Watermarks**
-To generate Watermarks whenever a certain event indicates that a new watermark can be generated, use the
-`AssignerWithPunctuatedWatermarks`. For this class, Flink will first call the `extractTimestamp(...)` method
-to assign the element a timestamp, and then immediately call for that element the
-`checkAndGetNextWatermark(...)` method.
+To generate watermarks whenever a certain event indicates that a new watermark might be generated, use
+`AssignerWithPunctuatedWatermarks`. For this class Flink will first call the `extractTimestamp(...)` method
+to assign the element a timestamp, and then immediately call the
+`checkAndGetNextWatermark(...)` method on that element.
-The `checkAndGetNextWatermark(...)` method gets the timestamp that was assigned in the `extractTimestamp(...)`
-method, and can decide whether it wants to generate a Watermark. Whenever the `checkAndGetNextWatermark(...)`
-method returns a non-null Watermark, and that Watermark is larger than the latest previous Watermark, that
-new Watermark will be emitted.
+The `checkAndGetNextWatermark(...)` method is passed the timestamp that was assigned in the `extractTimestamp(...)`
+method, and can decide whether it wants to generate a watermark. Whenever the `checkAndGetNextWatermark(...)`
+method returns a non-null watermark, and that watermark is larger than the latest previous watermark, that
+new watermark will be emitted.
<div class="codetabs" markdown="1">
<div data-lang="java" markdown="1">
@@ -326,4 +323,4 @@ class PunctuatedAssigner extends AssignerWithPunctuatedWatermarks[MyEvent] {
</div>
*Note:* It is possible to generate a watermark on every single event. However, because each watermark causes some
-computation downstream, an excessive number of watermarks slows down performance.
+computation downstream, an excessive number of watermarks degrades performance.