Flink on Zeppelin (3) - Streaming 篇

 关注
发布于: 2020 年 06 月 17 日
 继之前入门篇和Batch篇之后，今天这篇Flink on Zeppelin主要讲述如何在Zeppelin中使用Flink的Streaming功能，我们会以2个主要的场景来讲：
Streaming ETL
Streaming Data Analytics
准备工作﻿
      本文我们会用Kafka作为我们的数据源，使用Flink Sql处理Kafka中的某个topic数据，然后写入到另外一个Kafka Topic。为了使用Flink的Kafka connector，你需要在Flink Interpreter中配置flink.execution.packages。
flink.execution.packages org.apache.flink:flink-connector-kafka_2.11:1.10.0,org.apache.flink:flink-connector-kafka-base_2.11:1.10.0,org.apache.flink:flink-json:1.10.0
本文使用的kafka数据源是json格式，所以需要添加org.apache.flink:flink-json
       另外本文的例子会使用这个docker compose来创建Kafka Cluster，https://github.com/xushiyan/kafka-connect-datagen/
﻿
docker-compose up -d
curl -X POST http://localhost:8083/connectors \-H 'Content-Type:application/json' \-H 'Accept:application/json' \-d @connect.source.datagen.json
﻿
具体请参考这个官方链接 https://kafka-connect-datagen.readthedocs.io/en/latest/
﻿
Streaming ETL﻿
       接下里我们会用Flink SQL来做基于Kafka的Streaming ETL。首先我们需要创建Kafka source table代表kafka中的源数据。
%flink.ssql
DROP TABLE IF EXISTS source_kafka;
CREATE TABLE source_kafka (
    status  STRING,
    direction STRING,
    event_ts BIGINT
) WITH (
  'connector.type' = 'kafka',       
  'connector.version' = 'universal',
  'connector.topic' = 'generated.events',
  'connector.startup-mode' = 'earliest-offset',
  'connector.properties.zookeeper.connect' = 'localhost:2181',
  'connector.properties.bootstrap.servers' = 'localhost:9092',
  'connector.properties.group.id' = 'testGroup',
  'connector.startup-mode' = 'earliest-offset',
  'format.type'='json',
  'update-mode' = 'append'
);
﻿
然后创建Kafka sink table，代表清洗后的数据 （这里我们定义了WATERMARK，是为了下一步做基于window的流式数据分析）。
%flink.ssql
DROP TABLE IF EXISTS sink_kafka;
CREATE TABLE sink_kafka (
    status  STRING,
    direction STRING,
    event_ts TIMESTAMP(3),
    WATERMARK FOR event_ts AS event_ts - INTERVAL '5' SECOND
) WITH (
  'connector.type' = 'kafka',       
  'connector.version' = 'universal',    
  'connector.topic' = 'generated.events2',
  'connector.properties.zookeeper.connect' = 'localhost:2181',
  'connector.properties.bootstrap.servers' = 'localhost:9092',
  'connector.properties.group.id' = 'testGroup',
  'format.type'='json',
  'update-mode' = 'append'
)
﻿
接下里我们就可以用Insert Into语句来做Streaming ETL的工作了。
%flink.ssql
insert into sink_kafka select status, direction, cast(event_ts/1000000000 as timestamp(3)) from source_kafka where status <> 'foo'
﻿
这条Insert into语句非常简单，我们过滤掉了status为foo的数据，以及将event_ts 转化为timestamp类型。
﻿
﻿
然后可以用select语句来预览sink table中的数据来确认Streaming ETL正常工作。
﻿
Streaming Data Analytics﻿
       在完成了上面的Streaming ETL工作之后，我们就可以在Zeppelin中做流式数据分析了。在Zeppelin中可以用Select语句来做Flink 流数据分析，Select的结果会push到Zeppelin前端展示，可以用来做流式数据的dashboard。
﻿
Zeppelin支持3种模式的流式数据分析：
Single 模式 
Update 模式
Append 模式
Single模式﻿
%flink.ssql(type=single, parallelism=1, refreshInterval=3000, template=<h1>{1}</h1> until <h2>{0}</h2>)
select max(event_ts), count(1) from sink_kafka
﻿
﻿
Update模式        Update模式适合多行输出的情况，比如下面的select group by语句。这种模式会定期更新这多行数据，输出是Zeppelin的table格式，所以可以用Zeppelin自带的可视化控件。
﻿
%flink.ssql(type=update, refreshInterval=2000, parallelism=1)
select status, count(1) as pv from sink_kafka group by status
﻿
﻿
Append模式Append模式适合不断有新数据输出，但不会覆盖原有数据，只会不断append的情况。比如下面的基于窗口的group by语句。Append模式要求第一列数据类型是timestamp，这里的start_time就是timestamp类型。
﻿
%flink.ssql(type=append, parallelism=1, refreshInterval=2000, threshold=60000)
select TUMBLE_START(event_ts, INTERVAL '5' SECOND) as start_time, status, count(1) from sink_kafka
group by TUMBLE(event_ts, INTERVAL '5' SECOND), status
﻿
﻿
更多Flink SQL资料
本文只是简单介绍如何在Zeppelin中使用Flink Streaming SQL，关于更多Flink SQL请参考Flink官方文档
https://ci.apache.org/projects/flink/flink-docs-master/dev/table/sql/
https://ci.apache.org/projects/flink/flink-docs-master/dev/table/connect.html
﻿
Zeppelin on Flink （1） 入门篇
Zeppelin on Flink （2） Batch 篇
﻿
如果有碰到任何问题，请加入下面这个钉钉群讨论。后续我们会有更多Tutorial的文章，敬请期待。
﻿
﻿
Apache Zeppelin 公众号
﻿