You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ja...@apache.org on 2019/06/22 04:00:54 UTC
[flink] branch master updated: [docs-sync] Synchronize the latest
documentation changes (commits to 318ce571) into Chinese documents
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 00183f7 [docs-sync] Synchronize the latest documentation changes (commits to 318ce571) into Chinese documents
00183f7 is described below
commit 00183f7f904230b7fde18cc2dc692ddd4567e662
Author: Jark Wu <im...@gmail.com>
AuthorDate: Sat Jun 22 12:00:13 2019 +0800
[docs-sync] Synchronize the latest documentation changes (commits to 318ce571) into Chinese documents
---
docs/dev/api_concepts.zh.md | 7 +-
docs/dev/batch/index.zh.md | 6 +-
docs/dev/connectors/kafka.zh.md | 9 +
docs/dev/stream/operators/process_function.zh.md | 2 +-
docs/dev/stream/side_output.zh.md | 1 +
docs/dev/stream/state/checkpointing.zh.md | 6 +-
docs/dev/stream/state/state.zh.md | 44 ++-
docs/dev/stream/testing.zh.md | 416 ++++++++++++++++++-----
docs/dev/table/tableApi.zh.md | 185 +++++++---
docs/monitoring/metrics.zh.md | 6 +-
docs/ops/config.zh.md | 2 +-
docs/ops/deployment/aws.zh.md | 8 +-
docs/ops/state/checkpoints.zh.md | 2 +-
13 files changed, 536 insertions(+), 158 deletions(-)
diff --git a/docs/dev/api_concepts.zh.md b/docs/dev/api_concepts.zh.md
index 619ee70..8c2b746 100644
--- a/docs/dev/api_concepts.zh.md
+++ b/docs/dev/api_concepts.zh.md
@@ -694,7 +694,12 @@ Java and Scala classes are treated by Flink as a special POJO data type if they
- All fields are either public or must be accessible through getter and setter functions. For a field called `foo` the getter and setter methods must be named `getFoo()` and `setFoo()`.
-- The type of a field must be supported by Flink. At the moment, Flink uses [Avro](http://avro.apache.org) to serialize arbitrary objects (such as `Date`).
+- The type of a field must be supported by a registered serializer.
+
+POJOs are generally represented with a `PojoTypeInfo` and serialized with the `PojoSerializer` (using [Kryo](https://github.com/EsotericSoftware/kryo) as configurable fallback).
+The exception is when the POJOs are actually Avro types (Avro Specific Records) or produced as "Avro Reflect Types".
+In that case the POJO's are represented by an `AvroTypeInfo` and serialized with the `AvroSerializer`.
+You can also register your own custom serializer if required; see [Serialization](https://ci.apache.org/projects/flink/flink-docs-stable/dev/types_serialization.html#serialization-of-pojo-types) for further information.
Flink analyzes the structure of POJO types, i.e., it learns about the fields of a POJO. As a result POJO types are easier to use than general types. Moreover, Flink can process POJOs more efficiently than general types.
diff --git a/docs/dev/batch/index.zh.md b/docs/dev/batch/index.zh.md
index f097af9..6c567c8 100644
--- a/docs/dev/batch/index.zh.md
+++ b/docs/dev/batch/index.zh.md
@@ -829,7 +829,7 @@ File-based:
Collection-based:
-- `fromCollection(Collection)` - Creates a data set from the Java Java.util.Collection. All elements
+- `fromCollection(Collection)` - Creates a data set from a Java.util.Collection. All elements
in the collection must be of the same type.
- `fromCollection(Iterator, Class)` - Creates a data set from an iterator. The class specifies the
@@ -970,8 +970,8 @@ File-based:
Collection-based:
-- `fromCollection(Seq)` - Creates a data set from a Seq. All elements
- in the collection must be of the same type.
+- `fromCollection(Iterable)` - Creates a data set from an Iterable. All elements
+ returned by the Iterable must be of the same type.
- `fromCollection(Iterator)` - Creates a data set from an Iterator. The class specifies the
data type of the elements returned by the iterator.
diff --git a/docs/dev/connectors/kafka.zh.md b/docs/dev/connectors/kafka.zh.md
index c7d3f8d..40c03f1 100644
--- a/docs/dev/connectors/kafka.zh.md
+++ b/docs/dev/connectors/kafka.zh.md
@@ -129,6 +129,15 @@ The universal Kafka connector is compatible with older and newer Kafka brokers t
It is compatible with broker versions 0.11.0 or newer, depending on the features used.
For details on Kafka compatibility, please refer to the [Kafka documentation](https://kafka.apache.org/protocol.html#protocol_compatibility).
+### Migrating Kafka Connector from 0.11 to universal
+
+In order to perform the migration, see the [upgrading jobs and Flink versions guide]({{ site.baseurl }}/ops/upgrading.html)
+and:
+* Use Flink 1.9 or newer for the whole process.
+* Do not upgrade the Flink and operators at the same time.
+* Make sure that Kafka Consumer and/or Kafka Producer used in your job have assigned unique identifiers (`uid`):
+* Use stop with savepoint feature to take the savepoint (for example by using `stop --withSavepoint`)[CLI command]({{ site.baseurl }}/ops/cli.html).
+
### Usage
To use the universal Kafka connector add a dependency to it:
diff --git a/docs/dev/stream/operators/process_function.zh.md b/docs/dev/stream/operators/process_function.zh.md
index 081bcab..586238c 100644
--- a/docs/dev/stream/operators/process_function.zh.md
+++ b/docs/dev/stream/operators/process_function.zh.md
@@ -58,7 +58,7 @@ stream.keyBy(...).process(new MyProcessFunction())
## Low-level Joins
-To realize low-level operations on two inputs, applications can use `CoProcessFunction`. This
+To realize low-level operations on two inputs, applications can use `CoProcessFunction` or `KeyedCoProcessFunction`. This
function is bound to two different inputs and gets individual calls to `processElement1(...)` and
`processElement2(...)` for records from the two different inputs.
diff --git a/docs/dev/stream/side_output.zh.md b/docs/dev/stream/side_output.zh.md
index c7c4f3c..d20d3fe 100644
--- a/docs/dev/stream/side_output.zh.md
+++ b/docs/dev/stream/side_output.zh.md
@@ -60,6 +60,7 @@ Emitting data to a side output is possible from the following functions:
- [ProcessFunction]({{ site.baseurl }}/dev/stream/operators/process_function.html)
- [KeyedProcessFunction]({{ site.baseurl }}/dev/stream/operators/process_function.html#the-keyedprocessfunction)
- CoProcessFunction
+- KeyedCoProcessFunction
- [ProcessWindowFunction]({{ site.baseurl }}/dev/stream/operators/windows.html#processwindowfunction)
- ProcessAllWindowFunction
diff --git a/docs/dev/stream/state/checkpointing.zh.md b/docs/dev/stream/state/checkpointing.zh.md
index bcd32e0..fab1a49 100644
--- a/docs/dev/stream/state/checkpointing.zh.md
+++ b/docs/dev/stream/state/checkpointing.zh.md
@@ -102,6 +102,9 @@ env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
// enable externalized checkpoints which are retained after job cancellation
env.getCheckpointConfig().enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
+
+// allow job recovery fallback to checkpoint when there is a more recent savepoint
+env.getCheckpointConfig().setPreferCheckpointForRecovery(true);
{% endhighlight %}
</div>
<div data-lang="scala" markdown="1">
@@ -127,9 +130,6 @@ env.getCheckpointConfig.setFailTasksOnCheckpointingErrors(false)
// allow only one checkpoint to be in progress at the same time
env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)
-
-// allow job recovery fallback to checkpoint when there is a more recent savepoint
-env.getCheckpointConfig().setPreferCheckpointForRecovery(true)
{% endhighlight %}
</div>
</div>
diff --git a/docs/dev/stream/state/state.zh.md b/docs/dev/stream/state/state.zh.md
index 705b9db..78da641 100644
--- a/docs/dev/stream/state/state.zh.md
+++ b/docs/dev/stream/state/state.zh.md
@@ -343,6 +343,35 @@ val ttlConfig = StateTtlConfig
**注意:**
- 这种清理方式可以在任何时候通过 `StateTtlConfig` 启用或者关闭，比如在从 savepoint 恢复时。
+#### Cleanup in background
+
+Besides cleanup in full snapshot, you can also activate the cleanup in background. The following option
+will activate a default background cleanup in StateTtlConfig if it is supported for the used backend:
+
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+{% highlight java %}
+import org.apache.flink.api.common.state.StateTtlConfig;
+StateTtlConfig ttlConfig = StateTtlConfig
+ .newBuilder(Time.seconds(1))
+ .cleanupInBackground()
+ .build();
+{% endhighlight %}
+</div>
+ <div data-lang="scala" markdown="1">
+{% highlight scala %}
+import org.apache.flink.api.common.state.StateTtlConfig
+val ttlConfig = StateTtlConfig
+ .newBuilder(Time.seconds(1))
+ .cleanupInBackground
+ .build
+{% endhighlight %}
+</div>
+</div>
+
+For more fine-grained control over some special cleanup in background, you can configure it separately as described below.
+Currently, heap state backend relies on incremental cleanup and RocksDB backend uses compaction filter for background cleanup.
+
##### 增量数据清理
另外可以选择增量式清理状态数据，在状态访问或/和处理时进行。如果某个状态开启了该清理策略，则会在存储后端保留一个所有状态的惰性全局迭代器。
@@ -372,6 +401,7 @@ val ttlConfig = StateTtlConfig
</div>
该策略有两个参数。 第一个是每次清理时检查状态的条目数。如果启用，则始终按每个状态访问触发。第二个参数表示是否在处理每条记录时触发清理。
+If you enable the default background cleanup then this strategy will be activated for heap backend with 5 checked entries and without cleanup per record processing.
**注意:**
- 如果没有 state 访问，也没有处理数据，则不会清理过期数据。
@@ -395,7 +425,7 @@ import org.apache.flink.api.common.state.StateTtlConfig;
StateTtlConfig ttlConfig = StateTtlConfig
.newBuilder(Time.seconds(1))
- .cleanupInRocksdbCompactFilter()
+ .cleanupInRocksdbCompactFilter(1000)
.build();
{% endhighlight %}
</div>
@@ -406,15 +436,19 @@ import org.apache.flink.api.common.state.StateTtlConfig
val ttlConfig = StateTtlConfig
.newBuilder(Time.seconds(1))
- .cleanupInRocksdbCompactFilter
+ .cleanupInRocksdbCompactFilter(1000)
.build
{% endhighlight %}
</div>
</div>
-在 Flink 每处理固定数量（默认 1000）的状态后，RocksDB 压缩过滤器会使用当前时间戳进行是否过期的判断。可以通过
-`StateTtlConfig.newBuilder(...).cleanupInRocksdbCompactFilter(long queryTimeAfterNumEntries)` 方法进行定制。
-频繁的更新时间戳可以加快过期状态的速度，但是会降低压缩的效率，因为压缩会有 JNI 调用。
+RocksDB compaction filter will query current timestamp, used to check expiration, from Flink every time
+after processing certain number of state entries.
+You can change it and pass a custom value to
+`StateTtlConfig.newBuilder(...).cleanupInRocksdbCompactFilter(long queryTimeAfterNumEntries)` method.
+Updating the timestamp more often can improve cleanup speed
+but it decreases compaction performance because it uses JNI call from native code.
+If you enable the default background cleanup then this strategy will be activated for RocksDB backend and the current timestamp will be queried each time 1000 entries have been processed.
你还可以通过配置开启 RocksDB 过滤器的 debug 日志：
`log4j.logger.org.rocksdb.FlinkCompactionFilter=DEBUG`
diff --git a/docs/dev/stream/testing.zh.md b/docs/dev/stream/testing.zh.md
index 602713a..01c7808 100644
--- a/docs/dev/stream/testing.zh.md
+++ b/docs/dev/stream/testing.zh.md
@@ -23,25 +23,28 @@ specific language governing permissions and limitations
under the License.
-->
-This page briefly discusses how to test a Flink application in your IDE or a local environment.
+Testing is an integral part of every software development process as such Apache Flink comes with tooling to test your application code on multiple levels of the testing pyramid.
* This will be replaced by the TOC
{:toc}
-## Unit testing
+## Testing User-Defined Functions
-Usually, one can assume that Flink produces correct results outside of a user-defined `Function`. Therefore, it is recommended to test `Function` classes that contain the main business logic with unit tests as much as possible.
+Usually, one can assume that Flink produces correct results outside of a user-defined function. Therefore, it is recommended to test those classes that contain the main business logic with unit tests as much as possible.
-For example if one implements the following `ReduceFunction`:
+### Unit Testing Stateless, Timeless UDFs
+
+
+For example, let's take the following stateless `MapFunction`.
<div class="codetabs" markdown="1">
<div data-lang="java" markdown="1">
{% highlight java %}
-public class SumReduce implements ReduceFunction<Long> {
+public class IncrementMapFunction implements MapFunction<Long, Long> {
@Override
- public Long reduce(Long value1, Long value2) throws Exception {
- return value1 + value2;
+ public Long map(Long record) throws Exception {
+ return record +1 ;
}
}
{% endhighlight %}
@@ -49,30 +52,30 @@ public class SumReduce implements ReduceFunction<Long> {
<div data-lang="scala" markdown="1">
{% highlight scala %}
-class SumReduce extends ReduceFunction[Long] {
+class IncrementMapFunction extends MapFunction[Long, Long] {
- override def reduce(value1: java.lang.Long, value2: java.lang.Long): java.lang.Long = {
- value1 + value2
+ override def map(record: Long): Long = {
+ record + 1
}
}
{% endhighlight %}
</div>
</div>
-It is very easy to unit test it with your favorite framework by passing suitable arguments and verify the output:
+It is very easy to unit test such a function with your favorite testing framework by passing suitable arguments and verifying the output.
<div class="codetabs" markdown="1">
<div data-lang="java" markdown="1">
{% highlight java %}
-public class SumReduceTest {
+public class IncrementMapFunctionTest {
@Test
- public void testSum() throws Exception {
+ public void testIncrement() throws Exception {
// instantiate your function
- SumReduce sumReduce = new SumReduce();
+ IncrementMapFunction incrementer = new IncrementMapFunction();
// call the methods that you have implemented
- assertEquals(42L, sumReduce.reduce(40L, 2L));
+ assertEquals(3L, incrementer.map(2L));
}
}
{% endhighlight %}
@@ -80,44 +83,282 @@ public class SumReduceTest {
<div data-lang="scala" markdown="1">
{% highlight scala %}
-class SumReduceTest extends FlatSpec with Matchers {
+class IncrementMapFunctionTest extends FlatSpec with Matchers {
+
+ "IncrementMapFunction" should "increment values" in {
+ // instantiate your function
+ val incrementer: IncrementMapFunction = new IncrementMapFunction()
+
+ // call the methods that you have implemented
+ incremeter.map(2) should be (3)
+ }
+}
+{% endhighlight %}
+</div>
+</div>
+
+Similarly, a user-defined function which uses an `org.apache.flink.util.Collector` (e.g. a `FlatMapFunction` or `ProcessFunction`) can be easily tested by providing a mock object instead of a real collector. A `FlatMapFunction` with the same functionality as the `IncrementMapFunction` could be unit tested as follows.
+
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+{% highlight java %}
+public class IncrementFlatMapFunctionTest {
- "SumReduce" should "add values" in {
+ @Test
+ public void testIncrement() throws Exception {
// instantiate your function
- val sumReduce: SumReduce = new SumReduce()
+ IncrementFlatMapFunction incrementer = new IncrementFlatMapFunction();
+
+ Collector<Integer> collector = mock(Collector.class);
// call the methods that you have implemented
- sumReduce.reduce(40L, 2L) should be (42L)
+ incrementer.flatMap(2L, collector)
+
+ //verify collector was called with the right output
+ Mockito.verify(collector, times(1)).collect(3L);
}
}
{% endhighlight %}
</div>
+
+<div data-lang="scala" markdown="1">
+{% highlight scala %}
+class IncrementFlatMapFunctionTest extends FlatSpec with MockFactory {
+
+ "IncrementFlatMapFunction" should "increment values" in {
+ // instantiate your function
+ val incrementer : IncrementFlatMapFunction = new IncrementFlatMapFunction()
+
+ val collector = mock[Collector[Integer]]
+
+ //verify collector was called with the right output
+ (collector.collect _).expects(3)
+
+ // call the methods that you have implemented
+ flattenFunction.flatMap(2, collector)
+ }
+}
+{% endhighlight %}
+</div>
</div>
-## Integration testing
+### Unit Testing Stateful or Timely UDFs & Custom Operators
+
+Testing the functionality of a user-defined function, which makes use of managed state or timers is more difficult because it involves testing the interaction between the user code and Flink's runtime.
+For this Flink comes with a collection of so called test harnesses, which can be used to test such user-defined functions as well as custom operators:
-In order to end-to-end test Flink streaming pipelines, you can also write integration tests that are executed against a local Flink mini cluster.
+* `OneInputStreamOperatorTestHarness` (for operators on `DataStreams`s)
+* `KeyedOneInputStreamOperatorTestHarness` (for operators on `KeyedStream`s)
+* `TwoInputStreamOperatorTestHarness` (for operators of `ConnectedStreams` of two `DataStream`s)
+* `KeyedTwoInputStreamOperatorTestHarness` (for operators on `ConnectedStreams` of two `KeyedStream`s)
-In order to do so add the test dependency `flink-test-utils`:
+To use the test harnesses a set of additional dependencies (test scoped) is needed.
{% highlight xml %}
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-test-utils{{ site.scala_version_suffix }}</artifactId>
<version>{{site.version }}</version>
+ <scope>test</scope>
+</dependency>
+<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-runtime{{ site.scala_version_suffix }}</artifactId>
+ <version>{{site.version }}</version>
+ <scope>test</scope>
+ <classifier>tests</classifier>
+</dependency>
+<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-java{{ site.scala_version_suffix }}</artifactId>
+ <version>{{site.version }}</version>
+ <scope>test</scope>
+ <classifier>tests</classifier>
</dependency>
{% endhighlight %}
-For example, if you want to test the following `MapFunction`:
+Now, the test harnesses can be used to push records and watermarks into your user-defined functions or custom operators, control processing time and finally assert on the output of the operator (including side outputs).
<div class="codetabs" markdown="1">
<div data-lang="java" markdown="1">
{% highlight java %}
-public class MultiplyByTwo implements MapFunction<Long, Long> {
+
+public class StatefulFlatMapTest {
+ private OneInputStreamOperatorTestHarness<Long, Long> testHarness;
+ private StatefulFlatMap statefulFlatMapFunction;
+
+ @Before
+ public void setupTestHarness() throws Exception {
+
+ //instantiate user-defined function
+ statefulFlatMapFunction = new StatefulFlatMapFunction();
+
+ // wrap user defined function into a the corresponding operator
+ testHarness = new OneInputStreamOperatorTestHarness<>(new StreamFlatMap<>(statefulFlatMapFunction));
+
+ // optionally configured the execution environment
+ testHarness.getExecutionConfig().setAutoWatermarkInterval(50);
+
+ // open the test harness (will also call open() on RichFunctions)
+ testHarness.open();
+ }
+
+ @Test
+ public void testingStatefulFlatMapFunction() throws Exception {
+
+ //push (timestamped) elements into the operator (and hence user defined function)
+ testHarness.processElement(2L, 100L);
+
+ //trigger event time timers by advancing the event time of the operator with a watermark
+ testHarness.processWatermark(100L);
+
+ //trigger processing time timers by advancing the processing time of the operator directly
+ testHarness.setProcessingTime(100L);
+
+ //retrieve list of emitted records for assertions
+ assertThat(testHarness.getOutput(), containsInExactlyThisOrder(3L))
+
+ //retrieve list of records emitted to a specific side output for assertions (ProcessFunction only)
+ //assertThat(testHarness.getSideOutput(new OutputTag<>("invalidRecords")), hasSize(0))
+ }
+}
+
+{% endhighlight %}
+</div>
+
+<div data-lang="scala" markdown="1">
+{% highlight scala %}
+class StatefulFlatMapFunctionTest extends FlatSpec with Matchers with BeforeAndAfter {
+
+ private var testHarness: OneInputStreamOperatorTestHarness[Long, Long] = null
+ private var statefulFlatMap: StatefulFlatMapFunction = null
+
+ before {
+ //instantiate user-defined function
+ statefulFlatMap = new StatefulFlatMap
+
+ // wrap user defined function into a the corresponding operator
+ testHarness = new OneInputStreamOperatorTestHarness[Long, Long](new StreamFlatMap(statefulFlatMap))
+
+ // optionally configured the execution environment
+ testHarness.getExecutionConfig().setAutoWatermarkInterval(50);
+
+ // open the test harness (will also call open() on RichFunctions)
+ testHarness.open();
+ }
+
+ "StatefulFlatMap" should "do some fancy stuff with timers and state" in {
+
+
+ //push (timestamped) elements into the operator (and hence user defined function)
+ testHarness.processElement(2, 100);
+
+ //trigger event time timers by advancing the event time of the operator with a watermark
+ testHarness.processWatermark(100);
+
+ //trigger proccesign time timers by advancing the processing time of the operator directly
+ testHarness.setProcessingTime(100);
+
+ //retrieve list of emitted records for assertions
+ testHarness.getOutput should contain (3)
+
+ //retrieve list of records emitted to a specific side output for assertions (ProcessFunction only)
+ //testHarness.getSideOutput(new OutputTag[Int]("invalidRecords")) should have size 0
+ }
+}
+{% endhighlight %}
+</div>
+</div>
+
+`KeyedOneInputStreamOperatorTestHarness` and `KeyedTwoInputStreamOperatorTestHarness` are instantiated by additionally providing a `KeySelector` including `TypeInformation` for the class of the key.
+
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+{% highlight java %}
+
+public class StatefulFlatMapFunctionTest {
+ private OneInputStreamOperatorTestHarness<String, Long, Long> testHarness;
+ private StatefulFlatMap statefulFlatMapFunction;
+
+ @Before
+ public void setupTestHarness() throws Exception {
+
+ //instantiate user-defined function
+ statefulFlatMapFunction = new StatefulFlatMapFunction();
+
+ // wrap user defined function into a the corresponding operator
+ testHarness = new KeyedOneInputStreamOperatorTestHarness<>(new StreamFlatMap<>(statefulFlatMapFunction), new MyStringKeySelector(), Types.STRING);
+
+ // open the test harness (will also call open() on RichFunctions)
+ testHarness.open();
+ }
+
+ //tests
+
+}
+
+{% endhighlight %}
+</div>
+
+<div data-lang="scala" markdown="1">
+{% highlight scala %}
+class StatefulFlatMapTest extends FlatSpec with Matchers with BeforeAndAfter {
+
+ private var testHarness: OneInputStreamOperatorTestHarness[String, Long, Long] = null
+ private var statefulFlatMapFunction: FlattenFunction = null
+
+ before {
+ //instantiate user-defined function
+ statefulFlatMapFunction = new StateFulFlatMap
+
+ // wrap user defined function into a the corresponding operator
+ testHarness = new KeyedOneInputStreamOperatorTestHarness(new StreamFlatMap(statefulFlatMapFunction),new MyStringKeySelector(), Types.STRING())
+
+ // open the test harness (will also call open() on RichFunctions)
+ testHarness.open();
+ }
+
+ //tests
+
+}
+{% endhighlight %}
+</div>
+</div>
+
+Many more examples for the usage of these test harnesses can be found in the Flink code base, e.g.:
+
+* `org.apache.flink.streaming.runtime.operators.windowing.WindowOperatorTest` is a good example for testing operators and user-defined functions, which depend on processing or event time.
+* `org.apache.flink.streaming.api.functions.sink.filesystem.LocalStreamingFileSinkTest` shows how to test a custom sink with the `AbstractStreamOperatorTestHarness`. Specifically, it uses `AbstractStreamOperatorTestHarness.snapshot` and `AbstractStreamOperatorTestHarness.initializeState` to tests its interaction with Flink's checkpointing mechanism.
+
+<span class="label label-info">Note</span> Be aware that `AbstractStreamOperatorTestHarness` and its derived classes are currently not part of the public API and can be subject to change.
+
+## Testing Flink Jobs
+
+### JUnit Rule `MiniClusterWithClientResource`
+
+Apache Flink provides a JUnit rule called `MiniClusterWithClientResource` for testing complete jobs against a local, embedded mini cluster.
+called `MiniClusterWithClientResource`.
+
+To use `MiniClusterWithClientResource` one additional dependency (test scoped) is needed.
+
+{% highlight xml %}
+<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-test-utils{{ site.scala_version_suffix }}</artifactId>
+ <version>{{site.version }}</version>
+</dependency>
+{% endhighlight %}
+
+Let us take the same simple `MapFunction` as in the previous sections.
+
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+{% highlight java %}
+public class IncrementMapFunction implements MapFunction<Long, Long> {
@Override
- public Long map(Long value) throws Exception {
- return value * 2;
+ public Long map(Long record) throws Exception {
+ return record +1 ;
}
}
{% endhighlight %}
@@ -125,36 +366,44 @@ public class MultiplyByTwo implements MapFunction<Long, Long> {
<div data-lang="scala" markdown="1">
{% highlight scala %}
-class MultiplyByTwo extends MapFunction[Long, Long] {
+class IncrementMapFunction extends MapFunction[Long, Long] {
- override def map(value: Long): Long = {
- value * 2
+ override def map(record: Long): Long = {
+ record + 1
}
}
{% endhighlight %}
</div>
</div>
-You could write the following integration test:
+A simple pipeline using this `MapFunction` can now be tested in a local Flink cluster as follows.
<div class="codetabs" markdown="1">
<div data-lang="java" markdown="1">
{% highlight java %}
-public class ExampleIntegrationTest extends AbstractTestBase {
+public class ExampleIntegrationTest {
+
+ @ClassRule
+ public static MiniClusterWithClientResource flinkCluster =
+ new MiniClusterWithClientResource(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberSlotsPerTaskManager(2)
+ .setNumberTaskManagers(1)
+ .build());
@Test
- public void testMultiply() throws Exception {
+ public void testIncrementPipeline() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// configure your test environment
- env.setParallelism(1);
+ env.setParallelism(2);
// values are collected in a static variable
CollectSink.values.clear();
// create a stream of custom elements and apply transformations
env.fromElements(1L, 21L, 22L)
- .map(new MultiplyByTwo())
+ .map(new IncrementMapFunction())
.addSink(new CollectSink());
// execute
@@ -181,85 +430,76 @@ public class ExampleIntegrationTest extends AbstractTestBase {
<div data-lang="scala" markdown="1">
{% highlight scala %}
-class ExampleIntegrationTest extends AbstractTestBase {
+class StreamingJobIntegrationTest extends FlatSpec with Matchers with BeforeAndAfter {
- @Test
- def testMultiply(): Unit = {
- val env = StreamExecutionEnvironment.getExecutionEnvironment
+ val flinkCluster = new MiniClusterWithClientResource(new MiniClusterResourceConfiguration.Builder()
+ .setNumberSlotsPerTaskManager(1)
+ .setNumberTaskManagers(1)
+ .build)
- // configure your test environment
- env.setParallelism(1)
+ before {
+ flinkCluster.before()
+ }
- // values are collected in a static variable
- CollectSink.values.clear()
+ after {
+ flinkCluster.after()
+ }
- // create a stream of custom elements and apply transformations
- env
- .fromElements(1L, 21L, 22L)
- .map(new MultiplyByTwo())
- .addSink(new CollectSink())
- // execute
- env.execute()
+ "IncrementFlatMapFunction pipeline" should "incrementValues" in {
- // verify your results
- assertEquals(Lists.newArrayList(2L, 42L, 44L), CollectSink.values)
- }
-}
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+
+ // configure your test environment
+ env.setParallelism(2)
+
+ // values are collected in a static variable
+ CollectSink.values.clear()
+
+ // create a stream of custom elements and apply transformations
+ env.fromElements(1, 21, 22)
+ .map(new IncrementMapFunction())
+ .addSink(new CollectSink())
+
+ // execute
+ env.execute()
+ // verify your results
+ CollectSink.values should contain allOf (1,22,23)
+ }
+}
// create a testing sink
class CollectSink extends SinkFunction[Long] {
- override def invoke(value: java.lang.Long): Unit = {
- synchronized {
- values.add(value)
- }
+ override def invoke(value: Long): Unit = {
+ synchronized {
+ CollectSink.values.add(value)
}
+ }
}
object CollectSink {
-
// must be static
- val values: List[Long] = new ArrayList()
+ val values: util.List[Long] = new util.ArrayList()
}
{% endhighlight %}
</div>
</div>
-The static variable in `CollectSink` is used here because Flink serializes all operators before distributing them across a cluster.
-Communicating with operators instantiated by a local Flink mini cluster via static variables is one way around this issue.
-Alternatively, you could for example write the data to files in a temporary directory with your test sink.
-You can also implement your own custom sources for emitting watermarks.
-
-## Testing checkpointing and state handling
-
-One way to test state handling is to enable checkpointing in integration tests.
+A few remarks on integration testing with `MiniClusterWithClientResource`:
-You can do that by configuring your `StreamExecutionEnvironment` in the test:
+* In order not to copy your whole pipeline code from production to test, make sources and sinks pluggable in your production code and inject special test sources and test sinks in your tests.
-<div class="codetabs" markdown="1">
-<div data-lang="java" markdown="1">
-{% highlight java %}
-env.enableCheckpointing(500);
-env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 100));
-{% endhighlight %}
-</div>
-
-<div data-lang="scala" markdown="1">
-{% highlight scala %}
-env.enableCheckpointing(500)
-env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 100))
-{% endhighlight %}
-</div>
-</div>
+* The static variable in `CollectSink` is used here because Flink serializes all operators before distributing them across a cluster.
+Communicating with operators instantiated by a local Flink mini cluster via static variables is one way around this issue.
+Alternatively, you could write the data to files in a temporary directory with your test sink.
-And for example adding to your Flink application an identity mapper operator that will throw an exception
-once every `1000ms`. However writing such test could be tricky because of time dependencies between the actions.
+* You can implement a custom *parallel* source function for emitting watermarks if your job uses event timer timers.
-Another approach is to write a unit test using the Flink internal testing utility `AbstractStreamOperatorTestHarness` from the `flink-streaming-java` module.
+* It is recommended to always test your pipelines locally with a parallelism > 1 to identify bugs which only surface for the pipelines executed in parallel.
-For an example of how to do that please have a look at the `org.apache.flink.streaming.runtime.operators.windowing.WindowOperatorTest` also in the `flink-streaming-java` module.
+* Prefer `@ClassRule` over `@Rule` so that multiple tests can share the same Flink cluster. Doing so saves a significant amount of time since the startup and shutdown of Flink clusters usually dominate the execution time of the actual tests.
-Be aware that `AbstractStreamOperatorTestHarness` is currently not a part of public API and can be subject to change.
+* If your pipeline contains custom state handling, you can test its correctness by enabling checkpointing and restarting the job within the mini cluster. For this, you need to trigger a failure by throwing an exception from (a test-only) user-defined function in your pipeline.
{% top %}
diff --git a/docs/dev/table/tableApi.zh.md b/docs/dev/table/tableApi.zh.md
index acb1bb9..ddcdd5d 100644
--- a/docs/dev/table/tableApi.zh.md
+++ b/docs/dev/table/tableApi.zh.md
@@ -741,7 +741,7 @@ orders.groupBy("users").select("users, myUdagg.distinct(points) as myDistinctRes
Table orders = tableEnv.scan("Orders");
Table result = orders.distinct();
{% endhighlight %}
- <p><b>Note:</b> For streaming queries the required state to compute the query result might grow infinitely depending on the number of distinct fields. Please provide a query configuration with valid retention interval to prevent excessive state size. See <a href="streaming/query_configuration.html">Query Configuration</a> for details.</p>
+ <p><b>Note:</b> For streaming queries the required state to compute the query result might grow infinitely depending on the number of distinct fields. Please provide a query configuration with valid retention interval to prevent excessive state size. If state cleaning is enabled, distinct have to emit messages to prevent too early state eviction of downstream operators which makes distinct contains result updating. See <a href="streaming/query_configuration.html">Query Configurat [...]
</td>
</tr>
</tbody>
@@ -2636,56 +2636,93 @@ Table table = input
<tr>
<td>
- <strong>GroupBy TableAggregation</strong><br>
+ <strong>FlatAggregate</strong><br>
<span class="label label-primary">Streaming</span><br>
<span class="label label-info">Result Updating</span>
</td>
<td>
<p>Similar to a <b>GroupBy Aggregation</b>. Groups the rows on the grouping keys with the following running table aggregation operator to aggregate rows group-wise. The difference from an AggregateFunction is that TableAggregateFunction may return 0 or more records for a group. You have to close the "flatAggregate" with a select statement. And the select statement does not support aggregate functions.</p>
+ <p>Instead of using <code>emitValue</code> to output results, you can also use the <code>emitUpdateWithRetract</code> method. Different from <code>emitValue</code>, <code>emitUpdateWithRetract</code> is used to emit values that have been updated. This method outputs data incrementally in retract mode, i.e., once there is an update, we have to retract old records before sending new updated ones. The <code>emitUpdateWithRetract</code> method will be used in preference to the <code> [...]
{% highlight java %}
- public class MyMinMaxAcc {
- public int min = 0;
- public int max = 0;
+/**
+ * Accumulator for Top2.
+ */
+public class Top2Accum {
+ public Integer first;
+ public Integer second;
+}
+
+/**
+ * The top2 user-defined table aggregate function.
+ */
+public class Top2 extends TableAggregateFunction<Tuple2<Integer, Integer>, Top2Accum> {
+
+ @Override
+ public Top2Accum createAccumulator() {
+ Top2Accum acc = new Top2Accum();
+ acc.first = Integer.MIN_VALUE;
+ acc.second = Integer.MIN_VALUE;
+ return acc;
}
- public class MyMinMax extends TableAggregateFunction<Row, MyMinMaxAcc> {
- public void accumulate(MyMinMaxAcc acc, int value) {
- if (value < acc.min) {
- acc.min = value;
- }
- if (value > acc.max) {
- acc.max = value;
- }
+ public void accumulate(Top2Accum acc, Integer v) {
+ if (v > acc.first) {
+ acc.second = acc.first;
+ acc.first = v;
+ } else if (v > acc.second) {
+ acc.second = v;
}
+ }
- @Override
- public MyMinMaxAcc createAccumulator() {
- return new MyMinMaxAcc();
+ public void merge(Top2Accum acc, java.lang.Iterable<Top2Accum> iterable) {
+ for (Top2Accum otherAcc : iterable) {
+ accumulate(acc, otherAcc.first);
+ accumulate(acc, otherAcc.second);
}
+ }
- public void emitValue(MyMinMaxAcc acc, Collector<Row> out) {
- out.collect(Row.of(acc.min, acc.min));
- out.collect(Row.of(acc.max, acc.max));
+ public void emitValue(Top2Accum acc, Collector<Tuple2<Integer, Integer>> out) {
+ // emit the value and rank
+ if (acc.first != Integer.MIN_VALUE) {
+ out.collect(Tuple2.of(acc.first, 1));
}
-
- @Override
- public TypeInformation<Row> getResultType() {
- return new RowTypeInfo(Types.INT, Types.INT);
+ if (acc.second != Integer.MIN_VALUE) {
+ out.collect(Tuple2.of(acc.second, 2));
}
}
+}
-TableAggregateFunction tableAggFunc = new MyMinMax();
-tableEnv.registerFunction("myTableAggFunc", tableAggFunc);
+tEnv.registerFunction("top2", new Top2());
Table orders = tableEnv.scan("Orders");
Table result = orders
- .groupBy("a")
- .flatAggregate("myTableAggFunc(a) as (x, y)")
- .select("a, x, y");
+ .groupBy("key")
+ .flatAggregate("top2(a) as (v, rank)")
+ .select("key, v, rank");
{% endhighlight %}
<p><b>Note:</b> For streaming queries, the required state to compute the query result might grow infinitely depending on the type of aggregation and the number of distinct grouping keys. Please provide a query configuration with a valid retention interval to prevent excessive state size. See <a href="streaming/query_configuration.html">Query Configuration</a> for details.</p>
</td>
</tr>
+
+
+ <tr>
+ <td>
+ <strong>Group Window FlatAggregate</strong><br>
+ <span class="label label-primary">Streaming</span><br>
+ </td>
+ <td>
+ <p>Groups and aggregates a table on a <a href="#group-windows">group window</a> and possibly one or more grouping keys. You have to close the "flatAggregate" with a select statement. And the select statement does not support aggregate functions.</p>
+{% highlight java %}
+tableEnv.registerFunction("top2", new Top2());
+Table orders = tableEnv.scan("Orders");
+Table result = orders
+ .window(Tumble.over("5.minutes").on("rowtime").as("w")) // define window
+ .groupBy("a, w") // group by key and window
+ .flatAggregate("top2(b) as (v, rank)")
+ .select("a, w.start, w.end, w.rowtime, v, rank"); // access window properties and aggregate results
+{% endhighlight %}
+ </td>
+ </tr>
</tbody>
</table>
</div>
@@ -2804,53 +2841,97 @@ val table = input
<tr>
<td>
- <strong>GroupBy TableAggregation</strong><br>
+ <strong>FlatAggregate</strong><br>
<span class="label label-primary">Streaming</span><br>
<span class="label label-info">Result Updating</span>
</td>
<td>
<p>Similar to a <b>GroupBy Aggregation</b>. Groups the rows on the grouping keys with the following running table aggregation operator to aggregate rows group-wise. The difference from an AggregateFunction is that TableAggregateFunction may return 0 or more records for a group. You have to close the "flatAggregate" with a select statement. And the select statement does not support aggregate functions.</p>
+ <p>Instead of using <code>emitValue</code> to output results, you can also use the <code>emitUpdateWithRetract</code> method. Different from <code>emitValue</code>, <code>emitUpdateWithRetract</code> is used to emit values that have been updated. This method outputs data incrementally in retract mode, i.e., once there is an update, we have to retract old records before sending new updated ones. The <code>emitUpdateWithRetract</code> method will be used in preference to the <code> [...]
{% highlight scala %}
-case class MyMinMaxAcc(var min: Int, var max: Int)
+import java.lang.{Integer => JInteger}
+import org.apache.flink.table.api.Types
+import org.apache.flink.table.functions.TableAggregateFunction
+
+/**
+ * Accumulator for top2.
+ */
+class Top2Accum {
+ var first: JInteger = _
+ var second: JInteger = _
+}
-class MyMinMax extends TableAggregateFunction[Row, MyMinMaxAcc] {
+/**
+ * The top2 user-defined table aggregate function.
+ */
+class Top2 extends TableAggregateFunction[JTuple2[JInteger, JInteger], Top2Accum] {
- def accumulate(acc: MyMinMaxAcc, value: Int): Unit = {
- if (value < acc.min) {
- acc.min = value
- }
- if (value > acc.max) {
- acc.max = value
- }
+ override def createAccumulator(): Top2Accum = {
+ val acc = new Top2Accum
+ acc.first = Int.MinValue
+ acc.second = Int.MinValue
+ acc
}
- def resetAccumulator(acc: MyMinMaxAcc): Unit = {
- acc.min = 0
- acc.max = 0
+ def accumulate(acc: Top2Accum, v: Int) {
+ if (v > acc.first) {
+ acc.second = acc.first
+ acc.first = v
+ } else if (v > acc.second) {
+ acc.second = v
+ }
}
- override def createAccumulator(): MyMinMaxAcc = MyMinMaxAcc(0, 0)
-
- def emitValue(acc: MyMinMaxAcc, out: Collector[Row]): Unit = {
- out.collect(Row.of(Integer.valueOf(acc.min), Integer.valueOf(acc.min)))
- out.collect(Row.of(Integer.valueOf(acc.max), Integer.valueOf(acc.max)))
+ def merge(acc: Top2Accum, its: JIterable[Top2Accum]): Unit = {
+ val iter = its.iterator()
+ while (iter.hasNext) {
+ val top2 = iter.next()
+ accumulate(acc, top2.first)
+ accumulate(acc, top2.second)
+ }
}
- override def getResultType: TypeInformation[Row] = {
- new RowTypeInfo(Types.INT, Types.INT)
+ def emitValue(acc: Top2Accum, out: Collector[JTuple2[JInteger, JInteger]]): Unit = {
+ // emit the value and rank
+ if (acc.first != Int.MinValue) {
+ out.collect(JTuple2.of(acc.first, 1))
+ }
+ if (acc.second != Int.MinValue) {
+ out.collect(JTuple2.of(acc.second, 2))
+ }
}
}
-val tableAggFunc = new MyMinMax
+val top2 = new Top2
val orders: Table = tableEnv.scan("Orders")
val result = orders
- .groupBy('a)
- .flatAggregate(tableAggFunc('a) as ('x, 'y))
- .select('a, 'x, 'y)
+ .groupBy('key)
+ .flatAggregate(top2('a) as ('v, 'rank))
+ .select('key, 'v, 'rank)
{% endhighlight %}
<p><b>Note:</b> For streaming queries, the required state to compute the query result might grow infinitely depending on the type of aggregation and the number of distinct grouping keys. Please provide a query configuration with a valid retention interval to prevent excessive state size. See <a href="streaming/query_configuration.html">Query Configuration</a> for details.</p>
</td>
</tr>
+
+ <tr>
+ <td>
+ <strong>Group Window FlatAggregate</strong><br>
+ <span class="label label-primary">Streaming</span><br>
+ </td>
+ <td>
+ <p>Groups and aggregates a table on a <a href="#group-windows">group window</a> and possibly one or more grouping keys. You have to close the "flatAggregate" with a select statement. And the select statement does not support aggregate functions.</p>
+{% highlight scala %}
+val top2 = new Top2
+val orders: Table = tableEnv.scan("Orders")
+val result = orders
+ .window(Tumble over 5.minutes on 'rowtime as 'w) // define window
+ .groupBy('a, 'w) // group by key and window
+ .flatAggregate(top2('b) as ('v, 'rank))
+ .select('a, w.start, 'w.end, 'w.rowtime, 'v, 'rank) // access window properties and aggregate results
+
+{% endhighlight %}
+ </td>
+ </tr>
</tbody>
</table>
</div>
diff --git a/docs/monitoring/metrics.zh.md b/docs/monitoring/metrics.zh.md
index 1f14347..5dcd374 100644
--- a/docs/monitoring/metrics.zh.md
+++ b/docs/monitoring/metrics.zh.md
@@ -560,12 +560,14 @@ reporters will be instantiated on each job and task manager when they are starte
- `metrics.reporter.<name>.<config>`: Generic setting `<config>` for the reporter named `<name>`.
- `metrics.reporter.<name>.class`: The reporter class to use for the reporter named `<name>`.
+- `metrics.reporter.<name>.factory.class`: The reporter factory class to use for the reporter named `<name>`.
- `metrics.reporter.<name>.interval`: The reporter interval to use for the reporter named `<name>`.
- `metrics.reporter.<name>.scope.delimiter`: The delimiter to use for the identifier (default value use `metrics.scope.delimiter`) for the reporter named `<name>`.
- `metrics.reporters`: (optional) A comma-separated include list of reporter names. By default all configured reporters will be used.
-All reporters must at least have the `class` property, some allow specifying a reporting `interval`. Below,
-we will list more settings specific to each reporter.
+All reporters must at least have either the `class` or `factory.class` property. Which property may/should be used depends on the reporter implementation. See the individual reporter configuration sections for more information.
+Some reporters (referred to as `Scheduled`) allow specifying a reporting `interval`.
+Below more settings specific to each reporter will be listed.
Example reporter configuration that specifies multiple reporters:
diff --git a/docs/ops/config.zh.md b/docs/ops/config.zh.md
index a127d5c..cebdcf0 100644
--- a/docs/ops/config.zh.md
+++ b/docs/ops/config.zh.md
@@ -104,7 +104,7 @@ The default fraction for managed memory can be adjusted using the taskmanager.me
{% include generated/network_environment_configuration.html %}
-### Network communication (via Netty)
+### Network Communication (via Netty)
These parameters allow for advanced tuning. The default values are sufficient when running concurrent high-throughput jobs on a large cluster.
diff --git a/docs/ops/deployment/aws.zh.md b/docs/ops/deployment/aws.zh.md
index b465340..16e9bd7 100644
--- a/docs/ops/deployment/aws.zh.md
+++ b/docs/ops/deployment/aws.zh.md
@@ -56,7 +56,13 @@ when creating an EMR cluster.
After creating your cluster, you can [connect to the master node](http://docs.aws.amazon.com/ElasticMapReduce/latest/ManagementGuide/emr-connect-master-node.html) and install Flink:
1. Go the [Downloads Page]({{ site.download_url }}) and **download a binary version of Flink matching the Hadoop version** of your EMR cluster, e.g. Hadoop 2.7 for EMR releases 4.3.0, 4.4.0, or 4.5.0.
-2. Extract the Flink distribution and you are ready to deploy [Flink jobs via YARN](yarn_setup.html) after **setting the Hadoop config directory**:
+2. Make sure all the Hadoop dependencies are in the classpath before you submit any jobs to EMR:
+
+{% highlight bash %}
+export HADOOP_CLASSPATH=`hadoop classpath`
+{% endhighlight %}
+
+3. Extract the Flink distribution and you are ready to deploy [Flink jobs via YARN](yarn_setup.html) after **setting the Hadoop config directory**:
{% highlight bash %}
HADOOP_CONF_DIR=/etc/hadoop/conf ./bin/flink run -m yarn-cluster -yn 1 examples/streaming/WordCount.jar
diff --git a/docs/ops/state/checkpoints.zh.md b/docs/ops/state/checkpoints.zh.md
index c68bb0f..3b994a1 100644
--- a/docs/ops/state/checkpoints.zh.md
+++ b/docs/ops/state/checkpoints.zh.md
@@ -72,7 +72,7 @@ state.checkpoints.dir: hdfs:///checkpoints/
#### Configure for per job when constructing the state backend
{% highlight java %}
-env.setStateBackend(new RocksDBStateBackend("hdfs:///checkpoints-data/");
+env.setStateBackend(new RocksDBStateBackend("hdfs:///checkpoints-data/"));
{% endhighlight %}
### Difference to Savepoints