目录
因为在一篇博文上看到介绍“汽车之家介绍flink数据平台”中提到“基于 SQL 的开发流程”。基于kafka connector,通过source,sink,transformation三条sql完成数据接入,逻辑转换处理,结果落地三步工作。出于兴趣,自己去简(粗)单(糙)实现了这其中的一个小功能。相关的博文在这里,相关的代码上传到github。
简单说,通过kafka connector用3条sql实现如图所示功能:
但是实现的过程中也遇到了两个问题。
inStreamingMode
,并不支持inBatchMode
。不能实现汽车之家通过kafka connector来实现每日的定时统计。https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/connectors/table/kafka/
如图,只支持unbounded无界数据流,不支持bounded有界数据即batchmode.
在sink的时候是只支持append模式,而在append模式下,不支持group by,因为使用了group by 会改行结果行。
那么按照汽车之家每日统计PV,UV的需求必然需要使用到group by.按目前flink的最新版本也是办不到的。
如果强行使用group by 将会抛出异常:
目前sink只支持append模式,如果使用了group by 等会改变结果行,会报错:AppendStreamTableSink doesn't support consuming update changes which is produced by node GroupAggregate
对于spark和flink这种绝对主流的大数据框架,稍微上点规模的公司应该都有维护自己的内部分支,基于自家的业务做一些定制化开发。汽车之家应该也不例外。
所以以上功能flink社区版不能做到,汽车之家应该是基于内部的实现。
以上是背景介绍。
基于该功能并不算特别复杂,花了两天业余时间实现了。
要接入kafka,就要设置kafka连接信息,起始信息,以及结束信息
.
开始信息可选参数如下:
参数名
参数值
scan.startup.mode
可选值:'earliest-offset', 'latest-offset', 'group-offsets', 'timestamp' and 'specific-offsets'
scan.startup.specific-offsets
指定每个分区的偏移量,比如:'partition:0,offset:42;partition:1,offset:300'
scan.startup.timestamp-millis
直接指定开始时间戳,long类型
依葫芦画瓢,去除earliest-offset
,结束信息可选参数
可设置成:
参数名
参数值
scan.endup.mode
可选值:'latest-offset', 'group-offsets', 'timestamp' and 'specific-offsets'
scan.endup.specific-offsets
指定每个分区的偏移量,比如:'partition:0,offset:42;partition:1,offset:300'
scan.sendup.timestamp-millis
直接指定结束时间戳,long类型
这里涉及到一个版本的问题。flink kafka connector API在最近几个版本变化挺大的。就内部实现而言,1.13和1.14也有不小的变化。
比如,判断当前任务是否有界时,1.13版本是直接写为false
而在1.14版本变成了可动态判断并设置
public boolean isBounded() { return kafkaSource.getBoundedness() == Boundedness.BOUNDED; }
可以看到这里通过kafkaSource.getBoundedness()
获取当前任务是否有界,点进KafkaSource
,对于boundedness
属性,既然有getter
那必然有setter
啊。
果不其然,在KafkaSourceBuilder
类中提供了setBounded
方法。
这里还有意外惊喜,这个方法不但提供了设置bounded的功能,还能直接设置结束的参数。那么上一个kafka参数问题
解决了定义
问题,而在这里就解决了设置
的问题。
参数问题分为定义
和提交
。定义在第1部份已经解决,提交就是第2部份的setBounded
,但在哪里触发呢?
在KafkaDynamicSource
.createKafkaSource
的方法里。这里可仿照switch(startupMode)
写一个switch(endupuMode)
,在里面的分支去实现各种参数情况LATEST
,TIMESTAMP
,再在各个分支里设置kafka参数。
在case分支我们可以仿照kafkaSourceBuilder.setStartingOffsets
实现一个kafkaSourceBuilder.setEndOffsets
。在flink 1.13就得这么实现。但在flink 1.14,经过前面的分析得知setBounded
可设置结束参数。一举两得。
经过分析,我发现这个问题属于是庸人自扰。
AppendStreamTableSink doesn't support group by仅在streamingmode模式下,batchmode不存在改变结果行的问题,所以,只要改成了batchmode,天然的就不存在group by 异常问题。
选定flink 1.14版本,fork,拉取到本地,新建分支。
目前scan.endup.mode
只支持latest-offset
和timestamp
两种方式。
具体实现细节,就不一一贴代码了,有凑字数之嫌。
有兴趣实现细节的,可以查看这两个commit记录。大致就是这些改动。
完成代码:
代码实现完毕,本地编译。
使用maven,常规操作。有两个注意的点:
flink使用了spotless
进行代码格式化检测。修改了源码重新编译如果代码格式不对,可能就是没换行或者少了多了一个空格,就通过不了。
编译前,可以使用'mvn spotless:apply
自动校正。
flink 使用了Checkstyle
,一些代码使用了import static
,添加静态引入后进行编译时要注意。
编译成功后,可部署成单点或者伪集群模式测试。
这里采用本地测试。
flink-connector-kafka_2.11-1.14.0.jar
和flink-connector-kafka_2.11-1.14.0.xml
pom文件手动放入或者mvn install
本地仓库。确保将重新编译后的jar包引入项目
测试代码:
{
EnvironmentSettings fsSettings = EnvironmentSettings.newInstance()
.inBatchMode()
// .inStreamingMode()
.build();
TableEnvironment te = TableEnvironment.create(fsSettings);
String kafkaSql = "CREATE TABLE kafkatable (\n" +
" key STRING," +
" ts TIMESTAMP" +
") WITH (\n" +
" 'connector' = 'kafka',\n" +
" 'topic' = 'xxx',\n" +
" 'properties.bootstrap.servers' = 'xxx.xx.xxx.xxx:9092',\n" +
" 'properties.group.id' = 'xxx',\n" +
// -- optional: valid modes are "earliest-offset",
// -- "latest-offset", "group-offsets",
// -- or "specific-offsets"
" 'scan.startup.mode' = 'earliest-offset',\n" +
" 'scan.endup.mode' = 'latest-offset',\n" +
// " 'scan.endup.mode' = 'timestamp',\n" +
// " 'scan.endup.timestamp-millis' = '1641974234163',\n" +
// " 'scan.endup.mode' = 'latest-offset',\n" +
// "'scan.startup.specific-offsets' = 'partition:0,offset:20'," +
// " 'connector.specific-offsets.0.partition' = '0',"+
// "'connector.specific-offsets.0.offset' = '1',"+
" 'format' = 'json',\n" +
" 'json.fail-on-missing-field' = 'false',\n" +
" 'json.ignore-parse-errors' = 'true'\n" +
")";
te.executeSql(kafkaSql);
String sqlFile = "CREATE TABLE fs_table (\n" +
" dt VARCHAR,\n" +
" pv BIGINT,\n" +
" uv BIGINT" +
") WITH (\n" +
" 'connector'='filesystem',\n" +
" 'path'='d://path',\n" +
" 'format'='json',\n" +
" 'sink.partition-commit.delay'='1 s',\n" +
" 'sink.partition-commit.policy.kind'='success-file'\n" +
")";
te.executeSql(sqlFile);
te.executeSql("INSERT INTO fs_table\n" +
"SELECT\n" +
" 'as' as dt,\n" +
" COUNT(*) AS pv,\n" +
" COUNT(DISTINCT key) AS uv\n" +
"FROM kafkatable group by key\n").print();
}
测试代码的sql逻辑仅为测试。不要追究为什么count(*)是pv,随手写的。
测试代码
scan.endup.mode
设置为latest-offset
。如果要实现最开始的按天统计,如下设置。scan.startup.mode
同理。
" 'scan.endup.mode' = 'timestamp',\n" +
" 'scan.endup.timestamp-millis' = '1641974234163',\n" +
这段代码测试通过inBatchMode
引入kafka数据源,并将处理后的数据写入本地文件。
运行结果,测试通过。
完
手机扫一扫
移动阅读更方便
你可能感兴趣的文章