【flink番外篇】10、对有状态或及时 UDF 和自定义算子进行单元测试
•
大数据
Flink 系列文章
一、Flink 专栏
Flink 专栏系统介绍某一知识点，并辅以具体的示例进行说明。
-
1、Flink 部署系列
本部分介绍Flink的部署、配置相关基础内容。
-
2、Flink基础系列
本部分介绍Flink 的基础部分，比如术语、架构、编程模型、编程指南、基本的datastream api用法、四大基石等内容。
-
3、Flik Table API和SQL基础系列
本部分介绍Flink Table Api和SQL的基本用法，比如Table API和SQL创建库、表用法、查询、窗口函数、catalog等等内容。
-
4、Flik Table API和SQL提高与应用系列
本部分是table api 和sql的应用部分，和实际的生产应用联系更为密切，以及有一定开发难度的内容。
-
5、Flink 监控系列
本部分和实际的运维、监控工作相关。
二、Flink 示例专栏
Flink 示例专栏是 Flink 专栏的辅助说明，一般不会介绍知识点的信息，更多的是提供一个一个可以具体使用的示例。本专栏不再分目录，通过链接即可看出介绍的内容。
两专栏的所有文章入口点击：Flink 系列文章汇总索引
文章目录
- Flink 系列文章
- 一、有状态算子单元测试的介绍
-
- 1、DataStream API 测试依赖
- 2、Table API 测试依赖
- 3、maven依赖
- 二、flatmap function 单元测试
-
- 1、OneInputStreamOperatorTestHarness使用示例
- 2、KeyedOneInputStreamOperatorTestHarness使用示例
- 三、Process Function 单元测试
-
- 1、OneInputStreamOperatorTestHarness使用示例
- 2、ProcessFunctionTestHarnesses使用示例
本文着重介绍了Flink的有状态算子的单元测试，通过四个例子介绍了flatMap 和 process function的有状态单元测试。
如果需要了解更多内容，可以在本人Flink 专栏中了解更新系统的内容。
本文除了maven依赖外，没有其他依赖。
关于单元测试更详细内容参考：50、Flink的单元测试介绍及示例
一、有状态算子单元测试的介绍
对使用管理状态或定时器的用户自定义函数的功能测试会更加困难，因为它涉及到测试用户代码和 Flink 运行时的交互。 为此，Flink 提供了一组所谓的测试工具，可用于测试用户自定义函数和自定义算子：
- OneInputStreamOperatorTestHarness (适用于 DataStream 上的算子)
- KeyedOneInputStreamOperatorTestHarness (适用于 KeyedStream 上的算子)
- TwoInputStreamOperatorTestHarness (f适用于两个 DataStream 的 ConnectedStreams 算子)
- KeyedTwoInputStreamOperatorTestHarness (适用于两个 KeyedStream 上的 ConnectedStreams 算子)
要使用测试工具，还需要一组其他的依赖项，比如DataStream和TableAPI的依赖。
1、DataStream API 测试依赖
如果要为使用 DataStream API 构建的作业开发测试用例，则需要添加以下依赖项：
org.apache.flink
flink-test-utils
1.17.2
test
在各种测试实用程序中，该模块提供了 MiniCluster （一个可配置的轻量级 Flink 集群，能在 JUnit 测试中运行），可以直接执行作业。
2、Table API 测试依赖
如果您想在您的 IDE 中本地测试 Table API 和 SQL 程序，除了前述提到的 flink-test-utils 之外，您还要添加以下依赖项：
org.apache.flink
flink-table-test-utils
1.17.2
test
这将自动引入查询计划器和运行时，分别用于计划和执行查询。
flink-table-test-utils 模块已在 Flink 1.15 中引入，截至Flink 1.17版本被认为是实验性的。
3、maven依赖
本文示例的maven依赖
UTF-8
UTF-8
1.8
1.8
1.8
2.12
1.17.0
org.apache.flink
flink-clients
${flink.version}
provided
org.apache.flink
flink-java
${flink.version}
provided
org.apache.flink
flink-streaming-java
${flink.version}
provided
org.apache.flink
flink-csv
${flink.version}
provided
org.apache.flink
flink-json
${flink.version}
provided
junit
junit
4.13
org.mockito
mockito-core
4.0.0
test
二、flatmap function 单元测试
可以使用测试工具将记录和 watermark 推送到用户自定义函数或自定义算子中，控制处理时间，最后对算子的输出（包括旁路输出）进行校验。
1、OneInputStreamOperatorTestHarness使用示例
/*
* @Author: alanchan
* @LastEditors: alanchan
* @Description: 单元测试flatmap，如果是偶数则存储原值及平方数
*/
import java.util.concurrent.ConcurrentLinkedQueue;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.operators.StreamFlatMap;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.streaming.util.TestHarnessUtil;
import org.apache.flink.util.Collector;
import org.junit.Before;
import org.junit.Test;
public class TestStatefulFlatMapDemo3 {
static class AlanFlatMapFunction implements FlatMapFunction {
@Override
public void flatMap(Integer value, Collector out) throws Exception {
if (value % 2 == 0) {
out.collect(value);
out.collect(value * value);
}
}
}
OneInputStreamOperatorTestHarness testHarness;
@Before
public void setupTestHarness() throws Exception {
StreamFlatMap operator = new StreamFlatMap(new AlanFlatMapFunction());
testHarness = new OneInputStreamOperatorTestHarness(operator);
testHarness.open();
}
@Test
public void testFlatMap2() throws Exception {
long initialTime = 0L;
ConcurrentLinkedQueue expectedOutput = new ConcurrentLinkedQueue();
testHarness.processElement(new StreamRecord(1, initialTime + 1));
testHarness.processElement(new StreamRecord(2, initialTime + 2));
testHarness.processWatermark(new Watermark(initialTime + 2));
testHarness.processElement(new StreamRecord(3, initialTime + 3));
testHarness.processElement(new StreamRecord(4, initialTime + 4));
testHarness.processElement(new StreamRecord(5, initialTime + 5));
testHarness.processElement(new StreamRecord(6, initialTime + 6));
testHarness.processElement(new StreamRecord(7, initialTime + 7));
testHarness.processElement(new StreamRecord(8, initialTime + 8));
expectedOutput.add(new StreamRecord(2, initialTime + 2));
expectedOutput.add(new StreamRecord(4, initialTime + 2));
expectedOutput.add(new Watermark(initialTime + 2));
expectedOutput.add(new StreamRecord(4, initialTime + 4));
expectedOutput.add(new StreamRecord(16, initialTime + 4));
expectedOutput.add(new StreamRecord(6, initialTime + 6));
expectedOutput.add(new StreamRecord(36, initialTime + 6));
expectedOutput.add(new StreamRecord(8, initialTime + 8));
expectedOutput.add(new StreamRecord(64, initialTime + 8));
TestHarnessUtil.assertOutputEquals("输出结果", expectedOutput, testHarness.getOutput());
}
}
2、KeyedOneInputStreamOperatorTestHarness使用示例
KeyedOneInputStreamOperatorTestHarness 和 KeyedTwoInputStreamOperatorTestHarness 可以通过为键的类另外提供一个包含 TypeInformation 的 KeySelector 来实例化。
/*
* @Author: alanchan
* @LastEditors: alanchan
* @Description: 按照城市分类，并将城市缩写变成大写
*/
import com.google.common.collect.Lists;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.operators.StreamFlatMap;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.util.Collector;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
public class TestStatefulFlatMapDemo2 {
@Data
@NoArgsConstructor
@AllArgsConstructor
static class User {
private int id;
private String name;
private int age;
private String city;
}
static class AlanFlatMapFunction extends RichFlatMapFunction {
// The state is only accessible by functions applied on a {@code KeyedStream}
ValueState previousInput;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
previousInput = getRuntimeContext()
.getState(new ValueStateDescriptor("previousInput", User.class));
}
@Override
public void flatMap(User input, Collector out) throws Exception {
previousInput.update(input);
input.setCity(input.getCity().toUpperCase());
out.collect(input);
}
}
AlanFlatMapFunction alanFlatMapFunction = new AlanFlatMapFunction();
OneInputStreamOperatorTestHarness testHarness;
@Before
public void setupTestHarness() throws Exception {
alanFlatMapFunction = new AlanFlatMapFunction();
testHarness = new KeyedOneInputStreamOperatorTestHarness(new StreamFlatMap(alanFlatMapFunction),
new KeySelector() {
@Override
public String getKey(User value) throws Exception {
return value.getCity();
}
}, Types.STRING);
testHarness.open();
}
@Test
public void testFlatMap() throws Exception {
testHarness.processElement(new User(1, "alanchan", 18, "sh"), 10);
ValueState previousInput = alanFlatMapFunction.getRuntimeContext().getState(
new ValueStateDescriptor("previousInput", User.class));
User stateValue = previousInput.value();
Assert.assertEquals(
Lists.newArrayList(new StreamRecord(new User(1, "alanchan", 18, "sh".toUpperCase()), 10)),
testHarness.extractOutputStreamRecords());
Assert.assertEquals(new User(1, "alanchan", 18, "sh".toUpperCase()), stateValue);
testHarness.processElement(new User(2, "alan", 19, "bj"), 10000);
Assert.assertEquals(
Lists.newArrayList(
new StreamRecord(new User(1, "alanchan", 18, "sh".toUpperCase()), 10),
new StreamRecord(new User(2, "alan", 19, "bj".toUpperCase()), 10000)),
testHarness.extractOutputStreamRecords());
Assert.assertEquals(new User(2, "alan", 19, "bj".toUpperCase()), previousInput.value());
}
}
三、Process Function 单元测试
除了之前可以直接用于测试 ProcessFunction 的测试工具之外，Flink 还提供了一个名为 ProcessFunctionTestHarnesses 的测试工具工厂类，可以简化测试工具的实例化。
1、OneInputStreamOperatorTestHarness使用示例
import com.google.common.collect.Lists;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.operators.KeyedProcessOperator;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.util.Collector;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
/*
* @Author: alanchan
* @LastEditors: alanchan
* @Description:
*/
public class TestProcessOperatorDemo1 {
// public abstract class KeyedProcessFunction
static class AlanProcessFunction extends KeyedProcessFunction {
@Override
public void processElement(String value, KeyedProcessFunction.Context ctx,
Collector out) throws Exception {
ctx.timerService().registerProcessingTimeTimer(50);
out.collect("vx->" + value);
}
@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector out) throws Exception {
// 到达时间点触发事件操作
out.collect(String.format("定时器在 %d 被触发", timestamp));
}
}
private OneInputStreamOperatorTestHarness testHarness;
private AlanProcessFunction processFunction;
@Before
public void setupTestHarness() throws Exception {
processFunction = new AlanProcessFunction();
testHarness = new KeyedOneInputStreamOperatorTestHarness(
new KeyedProcessOperator(processFunction),
x -> "1",
Types.STRING);
// Function time is initialized to 0
testHarness.open();
}
@Test
public void testProcessElement() throws Exception {
testHarness.processElement("alanchanchn", 10);
Assert.assertEquals(
Lists.newArrayList(
new StreamRecord("vx->alanchanchn", 10)),
testHarness.extractOutputStreamRecords());
}
@Test
public void testOnTimer() throws Exception {
// test first record
testHarness.processElement("alanchanchn", 10);
Assert.assertEquals(1, testHarness.numProcessingTimeTimers());
// Function time 设置为 100
testHarness.setProcessingTime(100);
Assert.assertEquals(
Lists.newArrayList(
new StreamRecord("vx->alanchanchn", 10),
new StreamRecord("定时器在 100 被触发")),
testHarness.extractOutputStreamRecords());
}
}
2、ProcessFunctionTestHarnesses使用示例
本示例通过ProcessFunctionTestHarnesses验证了ProcessFunction、KeyedProcessFunction、CoProcessFunction、KeyedCoProcessFunction和BroadcastProcessFunction，基本完成了覆盖。
import java.util.Arrays;
import java.util.Collections;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.streaming.util.BroadcastOperatorTestHarness;
import org.apache.flink.streaming.util.KeyedTwoInputStreamOperatorTestHarness;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.streaming.util.ProcessFunctionTestHarnesses;
import org.apache.flink.streaming.util.TwoInputStreamOperatorTestHarness;
import org.apache.flink.util.Collector;
import org.junit.Assert;
import org.junit.Test;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
/*
* @Author: alanchan
*
* @LastEditors: alanchan
*
* @Description:
*/
public class TestProcessOperatorDemo3 {
@Data
@NoArgsConstructor
@AllArgsConstructor
static class User {
private int id;
private String name;
private int age;
private String city;
}
// 测试ProcessFunction 的 processElement
@Test
public void testProcessFunction() throws Exception {
// public abstract class ProcessFunction
ProcessFunction function = new ProcessFunction() {
@Override
public void processElement(
String value, Context ctx, Collector out) throws Exception {
out.collect("vx->" + value);
}
};
OneInputStreamOperatorTestHarness harness = ProcessFunctionTestHarnesses
.forProcessFunction(function);
harness.processElement("alanchanchn", 10);
Assert.assertEquals(harness.extractOutputValues(), Collections.singletonList("vx->alanchanchn"));
}
// 测试KeyedProcessFunction 的 processElement
@Test
public void testKeyedProcessFunction() throws Exception {
// public abstract class KeyedProcessFunction
KeyedProcessFunction function = new KeyedProcessFunction() {
@Override
public void processElement(String value, KeyedProcessFunction.Context ctx,
Collector out) throws Exception {
out.collect("vx->" + value);
}
};
OneInputStreamOperatorTestHarness harness = ProcessFunctionTestHarnesses
.forKeyedProcessFunction(function, x -> "name", BasicTypeInfo.STRING_TYPE_INFO);
harness.processElement("alanchan", 10);
Assert.assertEquals(harness.extractOutputValues(), Collections.singletonList(1));
}
// 测试CoProcessFunction 的 processElement1、processElement2
@Test
public void testCoProcessFunction() throws Exception {
// public abstract class CoProcessFunction
CoProcessFunction function = new CoProcessFunction() {
@Override
public void processElement1(String value, CoProcessFunction.Context ctx,
Collector out) throws Exception {
String[] userStr = value.split(",");
out.collect(
new User(Integer.parseInt(userStr[0]), userStr[1], Integer.parseInt(userStr[2]), userStr[3]));
}
@Override
public void processElement2(User value, CoProcessFunction.Context ctx,
Collector out) throws Exception {
out.collect(value);
}
};
TwoInputStreamOperatorTestHarness harness = ProcessFunctionTestHarnesses
.forCoProcessFunction(function);
harness.processElement2(new User(2, "alan", 19, "bj"), 100);
harness.processElement1("1,alanchan,18,sh", 10);
Assert.assertEquals(harness.extractOutputValues(),
Arrays.asList(new User(1, "alanchan", 18, "sh"), new User(2, "alan", 19, "bj")));
}
// 测试KeyedCoProcessFunction 的 processElement1和processElement2
@Test
public void testKeyedCoProcessFunction() throws Exception {
// public abstract class KeyedCoProcessFunction
KeyedCoProcessFunction function = new KeyedCoProcessFunction() {
@Override
public void processElement1(String value, KeyedCoProcessFunction.Context ctx,
Collector out) throws Exception {
String[] userStr = value.split(",");
out.collect(
new User(Integer.parseInt(userStr[0]), userStr[1], Integer.parseInt(userStr[2]), userStr[3]));
}
@Override
public void processElement2(User value, KeyedCoProcessFunction.Context ctx,
Collector out) throws Exception {
out.collect(value);
}
};
// public static
// KeyedTwoInputStreamOperatorTestHarness
// forKeyedCoProcessFunction(
// KeyedCoProcessFunction function,
// KeySelector keySelector1,
// KeySelector keySelector2,
// TypeInformation keyType)
KeyedTwoInputStreamOperatorTestHarness harness = ProcessFunctionTestHarnesses
.forKeyedCoProcessFunction(function, new KeySelector() {
@Override
public String getKey(String value) throws Exception {
return value.split(",")[3];
}
}, new KeySelector() {
@Override
public String getKey(User value) throws Exception {
return value.getCity();
}
}, TypeInformation.of(String.class));
harness.processElement2(new User(2, "alan", 19, "bj"), 100);
harness.processElement1("1,alanchan,18,sh", 10);
Assert.assertEquals(harness.extractOutputValues(),
Arrays.asList(new User(1, "alanchan", 18, "sh"), new User(2, "alan", 19, "bj")));
}
// 测试 BroadcastProcessFunction 的 processElement 和 processBroadcastElement
@Test
public void testBroadcastOperator() throws Exception {
// 定义广播
// 数据格式：
// sh,上海
// bj,北京
// public class MapStateDescriptor
MapStateDescriptor broadcastDesc = new MapStateDescriptor("Alan_RulesBroadcastState",
String.class,
String.class);
// public abstract class BroadcastProcessFunction
// * @param The input type of the non-broadcast side.
// * @param The input type of the broadcast side.
// * @param The output type of the operator.
BroadcastProcessFunction function = new BroadcastProcessFunction() {
// 负责处理广播流的元素
@Override
public void processBroadcastElement(String value, BroadcastProcessFunction.Context ctx,
Collector out) throws Exception {
System.out.println("收到广播数据：" + value);
// 得到广播流的存储状态
ctx.getBroadcastState(broadcastDesc).put(value.split(",")[0], value.split(",")[1]);
}
// 处理非广播流，关联维度
@Override
public void processElement(User value, BroadcastProcessFunction.ReadOnlyContext ctx,
Collector out) throws Exception {
// 得到广播流的存储状态
ReadOnlyBroadcastState state = ctx.getBroadcastState(broadcastDesc);
value.setCity(state.get(value.getCity()));
out.collect(value);
}
};
BroadcastOperatorTestHarness harness = ProcessFunctionTestHarnesses
.forBroadcastProcessFunction(function, broadcastDesc);
harness.processBroadcastElement("sh,上海", 10);
harness.processBroadcastElement("bj,北京", 20);
harness.processElement(new User(2, "alan", 19, "bj"), 10);
harness.processElement(new User(1, "alanchan", 18, "sh"), 30);
Assert.assertEquals(harness.extractOutputValues(),
Arrays.asList(new User(1, "alanchan", 18, "上海"), new User(2, "alan", 19, "北京")));
}
}
以上，本文着重介绍了Flink的有状态算子的单元测试，通过四个例子介绍了flatMap 和 process function的有状态单元测试。
本文来自网络，不代表协通编程立场，如若转载，请注明出处：https://www.net2asp.com/f3aa5bfde7.html
赞 (0)
RabbitMQ 重置用户名和密码的方法
« 上一篇
1天前
Flink的实时数据集成与ETL
下一篇 »
1天前
相关推荐
-
StarUML建模工具破解安装详细教程，保姆级教学！
StarUML建模工具安装破解详细教程 1.下载安装包 首先我们进入StarUml官网下载安装包 点击Downloads也可以选择自己想要的版本 2.安装StarUML 下载好后，…
-
分享24个网页游戏源代码，总有一个是你想要的
分享24个网页游戏源代码 24个游戏源代码下载链接：https://pan.baidu.com/s/1gYJlj8enJbh5mFS_wMaZBA?pwd=4ncb 提取码：4…
-
vue执行配置选项npm run serve的本质
vue执行配置选项npm run serve运行开发服务器的本质 目录 vue执行配置选项npm run serve运行开发服务器的本质 一、启动开发服务器、调用“工具链”、编译、…
-
Vue H5 前端自定义车牌号输入键盘(支持新能源)
项目介绍： 这个项目是基于 vue2+vant 写的录入车牌的键盘, 有需要的伙伴可以参考一下; 先看一下效果图: 这个是组件的代码,直接在页面中引用即可,车牌号的值存在ca…
-
前端工作经验总结以及技术分享
前端这个技术行业，吃的就是一门技术活，基础不牢，地动山摇，没有好的基础，就不能应对工作的提出的各种需求 博主不易，点个赞，关注一下再走啊（长期更新，下面列出几个常见的，后续详细的请…
-
vue 滚动条滚动到顶部或者底部
1:在滚动条所属 div 上加上 ref=“mianscroll” (mianscroll名称随意起) //中间内容 2:点击事件到顶部 topScrollClick() { th…
-
React的高阶组件详解
文章目录 React的高阶组件 高阶组件基本介绍 高阶组件使用过程 高阶组件应用场景 高阶组件的意义 React的高阶组件 高阶组件基本介绍 什么是高阶组件呢? 在认识高阶组价之前…
-
【表格单元格可编辑】vue-elementul简单实现table表格点击单元格可编辑，点击单元格变成弹框修改数据
前言 这是继我另一个帖子就是单元格点击变成输入框后添加的功能 因为考虑到有些时候修改单元格的信息可能点击后要修改很多，那一个输入框不好用 所以这时候就需要一个弹框可以把所有表单都显…
-
除夕最绚丽3D烟花代码（html+音效）
今天就是除夕了，新年怎么能少得了烟花呢？虽然绝大部分地区禁止燃放烟花，但该欣赏的烟花还是要欣赏滴~~ 最近整理文件，找到了一份烟花代码，3D特效，今天分享给大家，希望大家喜欢。 话…
-
TypeError: Failed to execute ‘createObjectURL‘ on ‘URL‘: Overload resolution failed
//request.js service.interceptors.response.use( (response) => { const res = response.da…
