java Flink（十五）Flink单元测试在Flink中进行单元测试 map、flatmap、状态、计时器等测试

野性酷女 2023-01-17 02:23 141阅读 0赞

今天记录一下flink单元测试的编写

flink中的单元测试模块也是基于JUnit来实现的，本文主要介绍部分方法用来测试flink中的富函数、状态函数（例如process）以及最简单的map、flatmap等

基本的JUnit操作建议百度学习。

首先回顾一下这几种flink函数

map是最基本的流操作，每次进来的流数据都会进行操作。

flatmap是可以有选择的进行操作、输出 输出需要用到Collector

状态函数例如process函数里边需要用到状态 状态用来保存之前的数据或者更新信息

计时器 例如定义一个时间或者事件，触发定时器执行各种操作

一、map 单元测试

编写源码：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70][]

代码：

import org.apache.flink.api.common.functions.MapFunction;
    
    public class MyStatelessMap implements MapFunction<String, String> {
        @Override
        public String map(String in) throws Exception {
            String out = "hello " + in;
            return out;
        }
    }

按快捷键 shift+control+t生成测试代码

测试代码：

import junit.framework.TestCase;
    import org.junit.Assert;
    import org.junit.Test;
    
    public class MyStatelessMapTest extends TestCase {
    
        @Test
        public void testMap() throws Exception {
            MyStatelessMap statelessMap = new MyStatelessMap();
            String out = statelessMap.map("world");
            System.out.println(out);
            Assert.assertEquals("hello world1", out);
        }
    }

二、flatmap 单元测试

flatmap我们进行单元测试需要传入collector参数

代码：

import org.apache.flink.api.common.functions.FlatMapFunction;
    import org.apache.flink.util.Collector;
    
    public class MyStatelessFlatMap implements FlatMapFunction<String, String> {
        @Override
        public void flatMap(String in, Collector<String> collector) throws Exception {
            String out = "hello " + in;
            collector.collect(out);
        }
    }

单元测试代码：

这里我们用flink提供的ListCollector来实现被测试代码中的collector参数。

import junit.framework.TestCase;
    import org.apache.flink.api.common.functions.util.ListCollector;
    import org.apache.flink.shaded.curator.org.apache.curator.shaded.com.google.common.collect.Lists;
    import org.junit.Assert;
    import org.junit.Test;
    
    import java.util.ArrayList;
    import java.util.List;
    
    public class MyStatelessFlatMapTest extends TestCase {
    
        @Test
        public void testFlatMap() throws Exception {
            MyStatelessFlatMap statelessFlatMap = new MyStatelessFlatMap();
            List<String> out = new ArrayList<>();
            ListCollector<String> listCollector = new ListCollector<>(out);
            statelessFlatMap.flatMap("world", listCollector);
            Assert.assertEquals(Lists.newArrayList("hello world"), out);
        }
    
    }

三、状态代码单元测试

我们平时的代码基本都是开窗的状态代码

代码：

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.configuration.Configuration;
    import org.apache.flink.util.Collector;
    
    public class StatefulFlatMap extends RichFlatMapFunction<String, String> {
        ValueState<String> previousInput;
    
        @Override
        public void open(Configuration parameters) throws Exception {
            previousInput = getRuntimeContext().getState(
                    new ValueStateDescriptor<String>("previousInput", Types.STRING));
        }
    
        @Override
        public void flatMap(String in, Collector<String> collector) throws Exception {
            String out = "hello " + in;
            if(previousInput.value() != null){
                out = out + " " + previousInput.value();
            }
            previousInput.update(in);
            collector.collect(out);
        }
    }

单元测试代码：

import junit.framework.TestCase;
    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.shaded.curator.org.apache.curator.shaded.com.google.common.collect.Lists;
    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.junit.Assert;
    import org.junit.Before;
    import org.junit.Test;
    
    public class StatefulFlatMapTest {
    
        @Test
        public void testFlatMap() throws Exception{
            StatefulFlatMap statefulFlatMap = new StatefulFlatMap();
    
            //模拟上下文生成
            OneInputStreamOperatorTestHarness<String, String> testHarness =
                    new KeyedOneInputStreamOperatorTestHarness<>(
                            new StreamFlatMap<>(statefulFlatMap), x -> "1", Types.STRING);
            testHarness.open();
    
            // test first record
            testHarness.processElement("world", 10);
            ValueState<String> previousInput =
                    statefulFlatMap.getRuntimeContext().getState(
                            new ValueStateDescriptor<>("previousInput", Types.STRING));
            String stateValue = previousInput.value();
            Assert.assertEquals(
                    Lists.newArrayList(new StreamRecord<>("hello world", 10)),
                    testHarness.extractOutputStreamRecords());
            Assert.assertEquals("world", stateValue);
    
            // test second record
            testHarness.processElement("parallel", 20);
            Assert.assertEquals(
                    Lists.newArrayList(
                            new StreamRecord<>("hello world", 10),
                            new StreamRecord<>("hello parallel world", 20)),
                    testHarness.extractOutputStreamRecords());
            Assert.assertEquals("parallel", previousInput.value());
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70]: /images/20221022/6ade4390111f437499d325a731921a11.png