MLSQL Stack 让流调试更加简单

前言

有一位同学正在调研MLSQL Stack对流的支持。然后说了流调试其实挺困难的。经过实践，希望实现如下三点：

• 能随时查看最新固定条数的Kafka数据
• 调试结果（sink）能打印在web控制台
• 流程序能自动推测json schema(现在spark是不行的)

实现这三个点之后，我发现调试确实就变得简单很多了。

流程

首先我新建了一个kaf_write.mlsql,里面方便我往Kafka里写数据：

set abc='''
{ "x": 100, "y": 200, "z": 200 ,"dataType":"A group"}
{ "x": 120, "y": 100, "z": 260 ,"dataType":"B group"}
{ "x": 120, "y": 100, "z": 260 ,"dataType":"B group"}
{ "x": 120, "y": 100, "z": 260 ,"dataType":"B group"}
{ "x": 120, "y": 100, "z": 260 ,"dataType":"B group"}
{ "x": 120, "y": 100, "z": 260 ,"dataType":"B group"}
{ "x": 120, "y": 100, "z": 260 ,"dataType":"B group"}
{ "x": 120, "y": 100, "z": 260 ,"dataType":"B group"}
{ "x": 120, "y": 100, "z": 260 ,"dataType":"B group"}
{ "x": 120, "y": 100, "z": 260 ,"dataType":"B group"}
{ "x": 120, "y": 100, "z": 260 ,"dataType":"B group"}
''';
load jsonStr.`abc` as table1;

select to_json(struct(*)) as value from table1 as table2;
save append table2 as kafka.`wow` where 
kafka.bootstrap.servers="127.0.0.1:9092";

这样我每次运行，数据就能写入到Kafka.

接着，我写完后，需要看看数据是不是真的都写进去了，写成了什么样子：

!kafkaTool sampleData 10 records from "127.0.0.1:9092" wow;

这句话表示，我要采样Kafka 10条Kafka数据，该Kafka的地址为 127.0.0.1:9092 ,主题为 wow .运行结果如下：

image.png

没有什么问题。接着我写一个非常简单的流式程序：

-- the stream name, should be uniq.
set streamName="streamExample";

-- use kafkaTool to infer schema from kafka
!kafkaTool registerSchema 2 records from "127.0.0.1:9092" wow;


load kafka.`wow` options 
kafka.bootstrap.servers="127.0.0.1:9092"
as newkafkatable1;


select *  from newkafkatable1
as table21;


-- print in webConsole instead of terminal console.
save append table21  
as webConsole.`` 
options mode="Append"
and duration="15"
and checkpointLocation="/tmp/s-cpl4";

运行结果如下：

image.png

在终端我们也可以看到实时效果了。