Flink中复杂拓扑(多输入)的集成测试

问题描述 投票:2回答:2

我需要为flink流式拓扑编写单元测试。它基本上是一个CoFlatMapFunction,它有2个输入。

我尝试从这个页面获得一些灵感:https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/testing.html

输入的顺序对我的拓扑很重要,所以当我测试时,我不能对每个输入使用StreamExecutionEnvironment#fromCollection,因为我不会控制在每个输入中注入数据点的顺序。

我尝试使用StreamExecutionEnvironment#fromCollection创建单个输入,并根据其类型将每个元素分派到我的CoFlatMapFunction的实际输入,但是在此操作中元素的顺序会丢失。

有没有其他方法来编写此测试?

java unit-testing apache-flink flink-streaming
2个回答
1
投票

flink训练练习有一个使用你可以参考的TwoInputStreamOperatorTestHarness的例子:

https://github.com/dataArtisans/flink-training-exercises/blob/master/src/test/java/com/dataartisans/flinktraining/exercises/datastream_java/process/EventTimeJoinTest.java

你需要这些依赖:

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-test-utils-junit</artifactId>
  <version>${flink.version}</version>
</dependency>

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-streaming-java_2.11</artifactId>
  <version>${flink.version}</version>
  <scope>test</scope>
  <type>test-jar</type>
</dependency>

<dependency>
  <groupId>org.mockito</groupId>
  <artifactId>mockito-all</artifactId>
  <version>1.10.19</version>
  <type>jar</type>
  <scope>test</scope>
</dependency>

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-runtime_2.11</artifactId>
  <version>${flink.version}</version>
  <scope>test</scope>
  <type>test-jar</type>
</dependency>

您应该记住,这不是一个公共的,受支持的界面,因此它可能以意想不到的方式发展。


1
投票

你想使用TwoInputStreamOperatorTestHarness类。不幸的是,文档有点稀疏。我有一个使用这个类的测试,但它还没有被推到133_stream-test-harnessflink-crawler分支。

© www.soinside.com 2019 - 2024. All rights reserved.