乐趣区

关于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

Hive Streaming 的意义

很多同学可能会好奇,为什么 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 & Dependency

因为只有在实现 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 的依赖也退出进去了。

写入 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 Streaming Sink

建一个 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 Streaming Source

因为 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。

Hive Temporal Table

看完了 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…

Note 下载:

https://github.com/lonelyGhos…

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

作者介绍:

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

退出移动版