大数据文摘出品
千呼万唤，Apache Flink 1.11.0终于上新了！
作为备受瞩目的新一代开源大数据计算引擎，Flink项目无疑已成为 Apache 基金会和 GitHub 最为活跃的项目之一。自2014年正式开源发展非常迅速，截止到2020年7月，社区的star数达到13600+，contributor 数达到718，有22989次commits，目前在GitHub上的访问量在Apache项目中位居前三。
作为快速发展的新一代大数据引擎，Flink本身的架构优势也吸引着越来越多的开源爱好者投入到社区的建设来。目前，Flink已经成为阿里巴巴、腾讯、字节跳动、滴滴、美团点评等知名公司建设流处理平台的首选。
不过你可能要问，流计算引擎这么多，为什么Flink这么受欢迎?
面对全量数据和增量数据，业界亟需用一套统一的大数据引擎技术来处理所有数据。
Apache Flink 被业界公认为最好的流计算引擎，其计算能力不仅仅局限于做流处理，而是一套兼具流、批、机器学习等多种计算功能的大数据引擎，用户只需根据业务逻辑开发一套代码，无论是全量数据还是增量数据，亦或者实时处理，一套方案即可全部支持。
Apache Flink 的系统架构
在近期发布的Apache Flink 1.11.0版本中，超过 200 名贡献者参与了 Flink 1.11.0 的开发，提交了超过 1300 个修复或优化。这些修改极大的提高了 Flink 的可用性，并且增强了各个 API 栈的功能。
先附上 GitHub 下载地址
https://flink.apache.org/downloads.html#apache-flink-1110
为了让大家更全面地了解 Apache Flink 背后的技术以及应用实践，大数据文摘联合Flink社区，为大家献上超值学习福利！马上下载，越早学习，越能抓住时代先机！
重磅福利1：Flink 年度学习资料大礼包！
这份大礼包里不仅有大数据实时计算及 Apache Flink 年度Flink 年度学习资料大礼包，还有300+页实战应用精华总结！
重磅福利2：阿里云《实时计算Flink极客训练营》免费开启！
不过，一个人看学习资料容易半途而废，配合这份“大礼包”，文摘菌在这里也安利一波阿里云下周开营的《实时计算Flink极客训练营》。
在5天的课程中你将获得：
点击下方链接，报名《实时计算Flink极客训练营》，并加入配套钉钉群，即可免费获取Flink年度学习大礼包！
https://developer.aliyun.com/learning/trainingcamp/sc/2?utm_content=g_1000181368
还有直播课程教你Flink正确打开方式，帮助你从 Flink 小白成长为 Flink 技术专家！
更重要的是，课程现在报名免费，名额有限，先到先得哦！
深度解读 Flink 1.11：流批一体 Hive 数仓
最新版的Flink1.11流计算结合 Hive 批处理数仓自上线以来就广受好评，我们最后也为大家带来了阿里云两位技术专家李劲松（之信）和李锐（天离）Flink 1.11流批一体 Hive 数仓的深度解读。
最新版的Flink 1.11 中流计算结合 Hive 批处理数仓，给离线数仓带来 Flink 流处理实时且 Exactly-once 的能力。另外，Flink 1.11 完善了 Flink 自身的 Filesystem connector，大大提高了 Flink 的易用性。
数仓架构
离线数仓
传统的离线数仓是由 Hive 加上 HDFS 的方案，Hive 数仓有着成熟和稳定的大数据分析能力，结合调度和上下游工具，构建一个完整的数据处理分析平台，流程如下：
这个流程下的问题是：
实时数仓
针对离线数仓的特点，随着实时计算的流行，越来越多的公司引入实时数仓，实时数仓基于 Kafka + Flink streaming，定义全流程的流计算作业，有着秒级甚至毫秒的实时性。
但是，实时数仓的一个问题是历史数据只有 3-15 天，无法在其上做 Ad-hoc 的查询。如果搭建 Lambda 的离线+实时的架构，维护成本、计算存储成本、一致性保证、重复的开发会带来很大的负担。
Hive 实时化
Flink 1.11 为解决离线数仓的问题，给 Hive 数仓带来了实时化的能力，加强各环节的实时性的同时，又不会给架构造成太大的负担。
Hive streaming sink
实时数据导入 Hive 数仓，你是怎么做的？Flume、Spark Streaming 还是 Flink Datastream？千呼万唤，Table / SQL 层的 streaming file sink 来啦，Flink 1.11 支持 Filesystem connector 和 Hive connector 的 streaming sink。
(注：图中 StreamingFileSink 的 Bucket 概念就是 Table/SQL 中的 Partition)
Table/SQL 层的 streaming sink 不仅：
而且引入了新的机制：Partition commit。
一个合理的数仓的数据导入，它不止包含数据文件的写入，也包含了 Partition 的可见性提交。当某个 Partition 完成写入时，需要通知 Hive metastore 或者在文件夹内添加 SUCCESS 文件。Flink 1.11 的 Partition commit 机制可以让你：
一个例子：
-- 结合Hive dialect使用Hive DDL语法
SET table.sql-dialect=hive;
CREATE TABLE hive_table (
user_id STRING,
order_amount DOUBLE
) PARTITIONED BY (
dt STRING,
hour STRING
) STORED AS PARQUET TBLPROPERTIES (
-- 使用partition中抽取时间，加上watermark决定partiton commit的时机
'sink.partition-commit.trigger'='partition-time',
-- 配置hour级别的partition时间抽取策略，这个例子中dt字段是yyyy-MM-dd格式的天，hour是0-23的小时，timestamp-pattern定义了如何从这两个partition字段推出完整的timestamp
'partition.time-extractor.timestamp-pattern'=’$dt $hour:00:00’,
-- 配置dalay为小时级，当 watermark > partition时间 + 1小时，会commit这个partition
'sink.partition-commit.delay'='1 h',
-- partitiion commit的策略是：先更新metastore(addPartition)，再写SUCCESS文件
'sink.partition-commit.policy.kind’='metastore,success-file'
SET table.sql-dialect=default;
CREATE TABLE kafka_table (
user_id STRING,
order_amount DOUBLE,
log_ts TIMESTAMP(3),
WATERMARK FOR log_ts AS log_ts - INTERVAL '5' SECOND
-- 可以结合Table Hints动态指定table properties [3]
INSERT INTO TABLE hive_table SELECT user_id, order_amount, DATE_FORMAT(log_ts, 'yyyy-MM-dd'), DATE_FORMAT(log_ts, 'HH') FROM kafka_table;
Hive streaming source
Hive 数仓中存在大量的 ETL 任务，这些任务往往是通过调度工具来周期性的运行，这样做主要有两个问题：
针对这些离线的 ETL 作业，Flink 1.11 为此开发了实时化的 Hive 流读，支持：
你甚至可以使用10分钟级别的分区策略，使用 Flink 的 Hive streaming source 和Hive streaming sink 可以大大提高 Hive 数仓的实时性到准实时分钟级，在实时化的同时，也支持针对 Table 全量的 Ad-hoc 查询，提高灵活性。
SELECT * FROM hive_table
/*+ OPTIONS('streaming-source.enable'=’true’,
'streaming-source.consume-start-offset'='2020-05-20') */;
实时数据关联 Hive 表
在 Flink 与 Hive 集成的功能发布以后，我们收到最多的用户反馈之一就是希望能够将 Flink 的实时数据与离线的 Hive 表进行关联。因此，在 Flink 1.11 中，我们支持将实时表与 Hive 表进行 temporal join。沿用 Flink 官方文档中的例子，假定 Orders 是实时表，而 LatestRates 是一张 Hive 表，用户可以通过以下语句进行temporal join：
SELECT
o.amout, o.currency, r.rate, o.amount * r.rate
FROM
Orders AS o
JOIN LatestRates FOR SYSTEM_TIME AS OF o.proctime AS r
ON r.currency = o.currency
与 Hive 表进行 temporal join 目前只支持 processing time，我们会把 Hive 表的数据缓存到内存中，并按照固定的时间间隔去更新缓存的数据。用户可以通过参数“lookup.join.cache.ttl” 来控制缓存更新的间隔，默认间隔为一个小时。
“lookup.join.cache.ttl” 需要配置到 Hive 表的 property 当中，因此每张表可以有不同的配置。另外，由于需要将整张 Hive 表加载到内存中，因此目前只适用于 Hive 表较小的场景。
Hive 增强
Hive Dialect 语法兼容
Flink on Hive 用户并不能很好的使用 DDL，主要是因为：
针对上述两个问题，我们提出了 FLIP-123，通过 Hive Dialect 为用户提供 Hive语法兼容。该功能的最终目标，是为用户提供近似 Hive CLI/Beeline 的使用体验，让用户无需在 Flink 和 Hive 的 CLI 之间进行切换，甚至可以直接迁移部分 Hive 脚本到 Flink 中执行。
在 Flink 1.11中，Hive Dialect 可以支持大部分常用的 DDL，比如 CREATE/ALTER TABLE、CHANGE/REPLACE COLUMN、ADD/DROP PARTITION 等等。为此，我们为 Hive Dialect 实现了一个独立的 parser，Flink 会根据用户指定的 Dialect 决定使用哪个 parser 来解析 SQL 语句。用户可以通过配置项“ table.sql-dialect ” 来指定使用的 SQL Dialect。它的默认值为 “default”，即 Flink 原生的 Dialect，而将其设置为 “hive” 时就开启了 Hive Dialect。对于 SQL 用户，可以在 yaml 文件中设置“table.sql-dialect” 来指定 session 的初始 Dialect，也可以通过 set 命令来动态调整需要使用的 Dialect，而无需重启 session。
Hive Dialect 目前所支持的具体功能可以参考 FLIP-123 或 Flink 的官方文档。另外，该功能的一些设计原则和使用注意事项如下：
向量化读取
Flink 1.10中，Flink 已经支持了 ORC (Hive 2+) 的向量化读取支持，但是这很局限，为此，Flink 1.11 增加了更多的向量化支持：
也就是说已经补全了所有版本的 Parquet 和 ORC 向量化支持，默认是开启的，提供开关。
简化 Hive 依赖
Flink 1.10 中，Flink 文档中列出了所需的 Hive 相关依赖，推荐用户自行下载。但是这仍然稍显麻烦，所以在1.11 中，Flink 提供了内置的依赖支持：
现在，你只需要单独下一个包，再搞定 HADOOP_CLASSPATH，即可运行 Flink on Hive。
Flink 增强
除了 Hive 相关的 features，Flink 1.11 也完成了大量其它关于流批一体的增强。
Flink Filesystem connector
Flink table 在长久以来只支持一个 csv 的 file system table，而且它还不支持Partition，行为上在某些方面也有些不符合大数据计算的直觉。
在 Flink 1.11，重构了整个 Filesystem connector 的实现 ：
例子：
CREATE TABLE fs_table (
user_id STRING,
order_amount DOUBLE,
dt STRING,
hour STRING
) PARTITIONED BY (dt, hour) WITH (
’connector’=’filesystem’,
’path’=’...’,
’format’=’parquet’,
'partition.time-extractor.timestamp-pattern'=’$dt $hour:00:00’,
'sink.partition-commit.delay'='1 h',
‘sink.partition-commit.policy.kind’='success-file')
-- stream environment or batch environment
INSERT INTO TABLE fs_table SELECT user_id, order_amount, DATE_FORMAT(log_ts, 'yyyy-MM-dd'), DATE_FORMAT(log_ts, 'HH') FROM kafka_table;
-- 通过 Partition 查询
SELECT * FROM fs_table WHERE dt=’2020-05-20’ and hour=’12’;
引入 Max Slot
Yarn perJob 或者 session 模式在 1.11 之前是无限扩张的，没有办法限制它的资源使用，只能用 Yarn queue 等方式来限制。但是传统的批作业其实都是大并发，运行在局限的资源上，一部分一部分阶段性的运行，为此，Flink 1.11 引入 Max Slot 的配置[11]，限制 Yarn application 的资源使用。
slotmanager.number-of-slots.max
定义 Flink 集群分配的最大 Slot 数。此配置选项用于限制批处理工作负载的资源消耗。不建议为流作业配置此选项，如果没有足够的 Slot，则流作业可能会失败。
想更进一步了解Flink？戳下方链接，报名《实时计算Flink极客训练营》，获取阿里云大数据独门绝学！
https://developer.aliyun.com/learning/trainingcamp/sc/2?utm_content=g_1000181368