关于flink:从-19-到-111聊聊-PyFlink-的核心功能演进附-Demo-代码

Flink 1.11 正式公布曾经三周了，其中最吸引我的个性就是 Hive Streaming。刚巧 Zeppelin-0.9-preview2 也在前不久公布了，所以就写了一篇 Zeppelin 上的 Flink Hive Streaming 的实战解析。本文次要从以下几局部跟大家分享：

Hive Streaming 的意义
Checkpoint & Dependency
写入 Kafka
Hive Streaming Sink
Hive Streaming Source
Hive Temporal Table

很多同学可能会好奇，为什么 Flink 1.11 中，Hive Streaming 的位置这么高？它的呈现，到底能给咱们带来什么？其实在大数据畛域，始终存在两种架构 Lambda 和 Kappa：

Lambda 架构——流批拆散，静态数据通过定时调度同步到 Hive 数仓，实时数据既会同步到 Hive，也会被实时计算引擎生产，这里就引出了一点问题。
数据口径问题
离线计算产出延时太大
数据冗余存储
Kappa 架构——全副应用实时计算来产出数据，历史数据通过回溯音讯的生产位点计算，同样也有很多的问题，毕竟没有一劳永逸的架构。
消息中间件无奈保留全副历史数据，同样数据都是行式存储，占用空间太大
实时计算计算历史数据力不从心
无奈进行 Ad-Hoc 的剖析

为了解决这些问题，行业内推出了实时数仓，解决了大部分痛点，然而还是有些中央力不从心。比方波及到历史数据的计算怎么办？我想做 Ad-Hoc 的剖析又怎么玩？所以行业内当初都是实时数仓与离线数仓并行存在，而这又带来了更多的问题：模型须要多份、数据产出不统一、历史数据的计算等等。

而 Hive Streaming 的呈现就能够解决这些问题！再也不必多套模型了；也不须要同一个指标因为波及到历史数据，写一遍实时 SQL 再写一遍离线 SQL；Ad-Hoc 也能做了，怎么做？读 Hive Streaming 产出的表就行！

接下来，让咱们从参数配置开始，接着流式的写入 Hive，再到流式的读取 Hive 表，最初再 Join 上 Hive 维表吧。这一整套流程都体验后，想必大家对 Hive Streaming 肯定会有更深刻的理解，更可能领会到它的作用。

因为只有在实现 Checkpoint 之后，文件才会从 In-progress 状态变成 Finish 状态，所以，咱们须要正当的去配置 Checkpoint，在 Zeppelin 中配置 Checkpoint 很简略。

%flink.conf

# checkpoint 配置

pipeline.time-characteristic EventTime
execution.checkpointing.interval 120000
execution.checkpointing.min-pause 60000
execution.checkpointing.timeout 60000
execution.checkpointing.externalized-checkpoint-retention RETAIN_ON_CANCELLATION

# 依赖 jar 包配置

flink.execution.packages org.apache.flink:flink-connector-kafka_2.11:1.11.0,org.apache.flink:flink-connector-kafka-base_2.11:1.11.0

又因为咱们须要从 Kafka 中读取数据，所以将 Kafka 的依赖也退出进去了。

咱们的数据来自于天池数据集，是以 CSV 的格局存在于本地磁盘，所以须要先将他们写入 Kafka。

先建一下 CSV Source 和 Kafka Sink 的表：

%flink.ssql
SET table.sql-dialect=default;
DROP TABLE IF EXISTS source_csv;
CREATE TABLE source_csv (
user_id string,
theme_id string,
item_id string,
leaf_cate_id string,
cate_level1_id string,
clk_cnt int,
reach_time string
) WITH (
 'connector' = 'filesystem',
 'path' = 'file:///Users/dijie/Downloads/Cloud_Theme_Click/theme_click_log.csv',
 'format' = 'csv'
 
 )

%flink.ssql
SET table.sql-dialect=default;
DROP TABLE IF EXISTS kafka_table;
CREATE TABLE kafka_table (
user_id string,
theme_id string,
item_id string,
leaf_cate_id string,
cate_level1_id string,
clk_cnt int,
reach_time string,
ts AS localtimestamp,
WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
) WITH (
'connector' = 'kafka',
'topic' = 'theme_click_log',
'properties.bootstrap.servers' = '10.70.98.1:9092',
'properties.group.id' = 'testGroup',
'format' = 'json',
'scan.startup.mode' = 'latest-offset'

)

因为注册的表即能够读又能够写，于是我在建表时将 Watermark 加上了；又因为源数据中的工夫戳曾经很老了，所以我这里采纳以后工夫减去 5 秒作为我的 Watermark。

大家能够看到，我在语句一开始指定了 SQL 方言为 Default，这是为啥呢？还有别的方言吗？别急，听我慢慢说。

其实在之前的版本，Flink 就曾经能够和 Hive 买通，包含能够把表建在 Hive 上，然而很多语法和 Hive 不兼容，包含建的表在 Hive 中也无奈查看，次要起因就是方言不兼容。所以，在 Flink 1.11 中，为了缩小学习老本（语法不兼容），能够用 DDL 建 Hive 表并在 Hive 中查问，Flink 反对了方言，默认的就是 Default 了，就和之前一样，如果想建 Hive 表，并反对查问，请应用 Hive 方言，具体能够参考下方链接。

Hive 方言：
https://ci.apache.org/project…

再把数据从 CSV 中读取后写入 Kafka。

%flink.ssql(type=update)

insert into kafka_table select * from source_csv ;

再瞄一眼 Kafka，看看数据有没有被灌进去：

看来没问题，那么接下来让咱们写入 Hive。

建一个 Hive Sink Table，记得将方言切换到 Hive，否则会有问题。

%flink.ssql
SET table.sql-dialect=hive;
DROP TABLE IF EXISTS hive_table;
CREATE TABLE hive_table (
user_id string,
theme_id string,
item_id string,
leaf_cate_id string,
cate_level1_id string,
clk_cnt int,
reach_time string
) PARTITIONED BY (dt string, hr string, mi string) STORED AS parquet TBLPROPERTIES (

 'partition.time-extractor.timestamp-pattern'='$dt $hr:$mi:00',
 'sink.partition-commit.trigger'='partition-time',
 'sink.partition-commit.delay'='1 min',
 'sink.partition-commit.policy.kind'='metastore,success-file'

);

参数给大家略微解释一下：

partition.time-extractor.timestamp-pattern：分区工夫抽取器，与 DDL 中的分区字段保持一致；
sink.partition-commit.trigger：分区触发器类型，可选 process-time 或 partition-time。process-time：不须要下面的参数，也不须要水印，当以后工夫大于分区创立工夫 +sink.partition-commit.delay 中定义的工夫，提交分区；partition-time：须要 Source 表中定义 watermark，当 watermark > 提取到的分区工夫 +sink.partition-commit.delay 中定义的工夫，提交分区；
sink.partition-commit.delay：相当于延时工夫；
sink.partition-commit.policy.kind：怎么提交，个别提交胜利之后，须要告诉 metastore，这样 Hive 能力读到你最新分区的数据；如果须要合并小文件，也能够自定义 Class，通过实现 PartitionCommitPolicy 接口。

接下来让咱们把数据插入刚刚创立的 Hive Table：

%flink.ssql

insert into hive_table select  user_id,theme_id,item_id,leaf_cate_id,cate_level1_id,clk_cnt,reach_time,DATE_FORMAT(ts, 'yyyy-MM-dd'), DATE_FORMAT(ts, 'HH') ,DATE_FORMAT(ts, 'mm') from kafka_table

让程序再跑一会儿~ 咱们先去倒一杯 95 年的 Java☕️。

而后再看看咱们的 HDFS，看看门路下的货色。

大家也能够用 Hive 自行查问看看，我呢就先卖个关子，一会儿用 Hive Streaming 来读数据。

因为 Hive 表下面曾经创立过了，所以这边读数据的时候间接拿来用就行了，不同的中央是须要应用 Table Hints 去笼罩参数。

Hive Streaming Source 最大的有余是，无奈读取曾经读取过的分区下新增的文件。简略来说就是，读过的分区，就不会再读了。看似很坑，不过认真想想，这样才合乎流的个性。

依旧给大家说一下参数的意思：

stream-source.enable：不言而喻，示意是否开启流模式。
stream-source.monitor-interval：监控新文件 / 分区产生的距离。
stream-source.consume-order：能够选 create-time 或者 partition-time；create-time 指的不是分区创立工夫，而是在 HDFS 中文件 / 文件夹的创立工夫；partition-time 指的是分区的工夫；对于非分区表，只能用 create-time。官网这边的介绍写的有点含糊，会让人误以为能够查到曾经读过的分区下新增的文件，其实通过我的测试和翻看源码发现并不能。
stream-source.consume-start-offset：示意从哪个分区开始读。

光说不干假把式，让咱们捞一把数据看看~

SET 那一行得带着，不然无奈应用 Table Hints。

看完了 Streaming Source 和 Streaming Sink，让咱们最初再试一下 Hive 作为维表吧。

其实用 Hive 维表很简略，只有是在 Hive 中存在的表，都能够当做维表应用，参数齐全能够用 Table Hints 来笼罩。

lookup.join.cache.ttl：示意缓存工夫；这里值得注意的是，因为 Hive 维表会把维表所有数据缓存在 TM 的内存中，如果维表量很大，那么很容易就 OOM；如果 ttl 工夫太短，那么会频繁的加载数据，性能会有很大影响。

因为是 LEFT JOIN，所以维表中不存在的数据会以 NULL 补全。再看一眼 DAG 图：

大家看一下画框的中央，能看到这边是应用的维表关联 LookupJoin。

如果大家 SQL 语句写错了，丢了 for system_time as of a.p，那么 DAG 图就会变成这样：

这种就不是维表 JOIN 其实更像是流和批在 JOIN。

Hive Streaming 的欠缺意味着买通了流批一体的最初一道壁垒，既能够做到历史数据的 OLAP 剖析，又能够实时吐出后果，这无疑是 ETL 开发者的福音，想必接下来的日子，会有更多的企业实现他们实时数仓的建设。

[1]https://ci.apache.org/project…
[2]https://github.com/apache/zep…

https://github.com/lonelyGhos…

最初，给大家介绍一下 Flink on Zeppelin 的钉钉群，大家有问题能够在外面探讨，Apache Zeppelin PMC 简锋大佬也在外面，有问题能够间接在钉群中发问交换～

狄杰，蘑菇街资深数据专家，负责蘑菇街实时计算平台。目前 Focus 在 Flink on Zeppelin，Apache Zeppelin Contributor。

关于flink:从-19-到-111聊聊-PyFlink-的核心功能演进附-Demo-代码

Hive Streaming 的意义

Checkpoint & Dependency

写入 Kafka

Hive Streaming Sink

Hive Streaming Source

Hive Temporal Table

写在最初

参考文档:

Note 下载:

作者介绍：