基于Java的Spark结构流单元测试

import org.apache.spark.streaming.dstream.DStream;
import org.apache.spark.streaming.TestInputStream;
import org.apache.spark.streaming.TestOutputStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.Seconds;

public class StreamingTest {
    public static void main(String[] args) throws InterruptedException {
        JavaStreamingContext jssc = new JavaStreamingContext("local[2]", "TestStreaming", Seconds(1));

        // 创建一个模拟输入流
        TestInputStream<String> testInputStream = new TestInputStream<>(jssc.ssc(), new String[]{"hello", "world"});

        // 将模拟输入流转换为 DStream
        JavaReceiverInputDStream<String> lines = jssc.receiverStream(testInputStream);

        // 处理逻辑
        DStream<String> words = lines.flatMap(x -> Arrays.asList(x.split(" ")).iterator());
        TestOutputStream<String> testOutputStream = new TestOutputStream<>(words, new String[0]);
        testOutputStream.register();

        // 启动 StreamingContext
        jssc.start();
        jssc.awaitTerminationOrTimeout(3000);
    }
}

import org.apache.spark.streaming.TestOutputStream;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.Seconds;
import org.junit.Assert;
import org.junit.Test;

public class StreamingTest {
    @Test
    public void testStreamingOutput() throws InterruptedException {
        JavaStreamingContext jssc = new JavaStreamingContext("local[2]", "TestStreaming", Seconds(1));

        // 创建一个模拟输入流
        TestInputStream<String> testInputStream = new TestInputStream<>(jssc.ssc(), new String[]{"hello", "world"});

        // 将模拟输入流转换为 DStream
        JavaReceiverInputDStream<String> lines = jssc.receiverStream(testInputStream);

        // 处理逻辑
        JavaDStream<String> words = lines.flatMap(x -> Arrays.asList(x.split(" ")).iterator());

        // 捕获输出结果
        TestOutputStream<String> testOutputStream = new TestOutputStream<>(words, new String[0]);
        testOutputStream.register();

        // 启动 StreamingContext
        jssc.start();
        jssc.awaitTerminationOrTimeout(3000);

        // 验证输出结果
        String[] output = testOutputStream.getOutput().get(0);
        Assert.assertArrayEquals(new String[]{"hello", "world"}, output);
    }
}