java中的spark结构化流单元测试

hivapdat  于 2021-05-17  发布在  Spark
关注(0)|答案(1)|浏览(507)

我正在开发一个api,使用spark结构化流技术从kafka读取数据并写入java中的blob存储。我找不到为此编写单元测试的方法。我有一个reader类返回一个数据集,还有一个writer类将数据集作为输入,并以指定的格式写入blob存储。我在memorystream上看到了一些博客,但我觉得这还不够。
提前谢谢。

ecfsfe2w

ecfsfe2w1#

显然,您可以参考这个答案,了解如何使用内存流进行单元测试-单元测试-结构化流
另外,你可以看看霍顿卡劳的Spark测试基地。Spark试验基地
您可以模拟来自kafka的流Dataframe,并在该Dataframe之上的代码中为转换运行测试用例。
样品:

static Dataset<Row> createTestStreamingDataFrame() {
    MemoryStream<String> testStream= new MemoryStream<String>(100, sqlContext(), Encoders.STRING());
    testStream.addData((Arrays.asList("1,1","2,2","3,3")).toSeq());
    return testStream.toDF().selectExpr(
        "cast(split(value,'[,]')[0] as int) as testCol1",
        "cast(split(value,'[,]')[1] as int) as testCol2");
}

相关问题