调研FlinkSql功能测试及实战演练
【调研】FlinkSql功能测试及实战演练
槐序i 2021-05-12 17:39:24 573 收藏
分类专栏： flink
版权
flink
专栏收录该内容
14 篇⽂章0 订阅
订阅专栏
FlinkSql功能测试及实战演练
前⾔：Flink在国内的占有率逐步提升的情况下，各项组件的功能与稳定性也得到逐步提升。为了解决⽬
前已有的复杂需求，尝试研究flinksql的特性与功能，作为是否引⼊该组件的依据。同时尝试将现有需求通过简单demo的形式进⾏测试。本次测试主要集中在Kafka、mysql、Impala三个组件上，同时将结合官⽅⽂档进⾏：
/projects/flink/flink-docs-release-1.12/zh/
1
1 前期准备
1.1 环境配置
本次研究测试需要⽤到以下组件：
CDH 6.3.2
Flink 1.12.2
mysql 5.7
impala 3.2.0-cdh6.3.2
kafka 2.2.1-cdh6.3.2
1
2
3
4
5
1.2 依赖关系
本次测试会将FlinkSql与kafka、mysql、impala等组件进⾏conn，因此需要以下依赖包：
flink-connector-kafka_2.11-1.12.2.jar
flink-connector-jdbc_2.11-1.11.2.jar
mysql-connector-java-5.1.47.jar
ImpalaJDBC4.jar
ImpalaJDBC41.jar
flink-sql-connector-kafka_2.11-1.12.2.jar
1
2
3
4
5
6
1.3 重启flink
将上述所需的jar包放⼊$FLINK_HOME/lib中之后（所有部署flink的服务器都需要放），重启yarn-session
yarn-session.sh --detached
sql-client.sh embedded
1
2
2 FlinkSql-kafka测试
FlinkSql-kafka相关资料：
/projects/flink/flink-docs-release-1.12/zh/dev/table/connectors/kafka.html
1
2.1 FlinkSql-kafka常规功能测试
通过FlinkSql将Kafka中的数据映射成⼀张表
2.1.1 创建常规topic
1、创建topic
kafka-topics --create --zookeeper 192.168.5.185:2181,192.168.5.165:2181,192.168.5.187:2181 --replication-factor 3 --partitions 3 --topic test01
2、模拟消费者
kafka-console-consumer --bootstrap-server 192.168.5.185:9092,192.168.5.165:9092,192.168.5.187:9092 --topic
test01 --from-beginning
3、模拟⽣产者
kafka-console-producer --broker-list 192.168.5.185:9092,192.168.5.165:9092,192.168.5.187:9092 --topic test01
kafka-topics --delete --topic test01 --zookeeper 192.168.5.185:2181,192.168.5.165:2181,192.168.5.187:2181
1
2
3
4
5
mysql操作官方文档
6
7
8
9
10
11
2.1.2 FlinkSql建表
CREATE TABLE t1 (
name string,
age BIGINT,
isStu INT,
opt STRING,
optDate TIMESTAMP(3) METADATA FROM ‘timestamp’
) WITH (
‘connector’ = ‘kafka’, – 使⽤ kafka connector
‘topic’ = ‘test01’, – kafka topic
‘de’ = ‘earliest-offset’,
‘properties.bootstrap.servers’ = ‘192.168.5.185:9092,192.168.5.165:9092,192.168.5.187:9092’, – kafka broker 地址
‘format’ = ‘csv’ – 数据源格式为 csv，
);
select * from t1;
1
2
3
4
5
6
7
8
9
10
11
12
13
14
2.1.3 写⼊数据
往kafka中写⼊数据，同时查看flinksql中t1表的变化
zhangsan,20,1,1
lisi,18,1,2
wangwu,30,2,2
1
2
3
通过kafka数据映射成表这个步骤，可以将数据实时的汇⼊表中，通过sql再进⾏后续操作，相对代码编写来说更为简易，同时有问题也相对好排查
2.2 FlinkSql-upsertKafka常规功能测试
upsert-kafka 连接器⽣产 changelog 流，其中每条数据记录代表⼀个更新或删除事件。
如果有key则update，没有key则insert，如果value的值为空，则表⽰删除
2.2.1 FlinkSql建upsert表
drop table t2;
CREATE TABLE t2 (
name STRING,
age bigint,
isStu INT,
opt STRING,
optDate TIMESTAMP(3) ,
PRIMARY KEY (name) NOT ENFORCED
) WITH (
‘connector’ = ‘upsert-kafka’,
‘topic’ = ‘test02’,
‘properties.bootstrap.servers’ = ‘192.168.5.185:9092,192.168.5.165:9092,192.168.5.187:9092’, – kafka broker 地址
‘key.format’ = ‘csv’,
‘value.format’ = ‘csv’
);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
2.2.2 建⽴映射关系
将t1表中的数据写⼊到t2中
INSERT INTO t2 SELECT * FROM t1 ;
select * from t2;
1
2
结果如下：
2.2.3 更新数据
继续模拟kafka⽣产者，写⼊如下数据
zhangsan,25,1,2
risen,8,8,8
lisi,0,0,
1
2
3
结果如下：
2.2.4⼩结
通过如上测试，两条更新，⼀条插⼊，都已经实现了，
根据官⽅⽂档描述，指定key的情况下，当value为空则判断为删除操作
但是假如我插⼊⼀条数据到kafka，例如：
lisi,
1
只有key，没有value，t1表就会报如下错误
因为建表的时候有⼏个类型定义为了Int类型，这⾥为空它默认为是""空字符串，有点呆，推测如果是json格式这类可以指定数据类型的，才能直接使⽤。对于csv这种数据类型不确定的，会存在⽆法推断类型的情况。
鉴于此，为了探究是否真的具备删除操作，我⼜将上述所有表结构都进⾏了修改。为了试验简单，我直接修改表结构再次测试
drop TABLE t1;
CREATE TABLE t1 (
name STRING,
age STRING,
isStu STRING,
opt STRING,
optDate TIMESTAMP(3) METADATA FROM ‘timestamp’
) WITH (
‘connector’ = ‘kafka’, – 使⽤ kafka connector
‘topic’ = ‘test01’, – kafka topic
‘de’ = ‘earliest-offset’,
‘properties.bootstrap.servers’ = ‘192.168.5.185:9092,192.168.5.165:9092,192.168.5.187:9092’, – kafka broker 地址
‘format’ = ‘csv’ – 数据源格式为 csv，
);
drop table t2;
CREATE TABLE t2 (
name STRING,
age STRING,
isStu STRING,
opt STRING,
optDate TIMESTAMP(3) ,
PRIMARY KEY (name) NOT ENFORCED
) WITH (
‘connector’ = ‘upsert-kafka’,
‘topic’ = ‘test02’,
‘properties.bootstrap.servers’ = ‘192.168.5.185:9092,192.168.5.165:9092,192.168.5.187:9092’, – kafka broker 地址
‘key.format’ = ‘csv’,
‘value.format’ = ‘csv’
);
INSERT INTO t2 SELECT * FROM t1 ;
select * from t2;
1