Flink Batch SQL 1.10 实践

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介: 1.10可以说是第一个成熟的生产可用的Flink Batch SQL版本,它一扫之前Dataset的羸弱,从功能和性能上都有大幅改进,以下我从架构、外部系统集成、实践三个方面进行阐述。

作者:李劲松(之信)

Flink作为流批统一的计算框架,在1.10中完成了大量batch相关的增强与改进。1.10可以说是第一个成熟的生产可用的Flink Batch SQL版本,它一扫之前Dataset的羸弱,从功能和性能上都有大幅改进,以下我从架构、外部系统集成、实践三个方面进行阐述。

架构

Stack

图片 1.png

首先来看下stack,在新的Blink planner中,batch也是架设在Transformation上的,这就意味着我们和Dataset完全没有关系了:

  1. 我们可以尽可能的和streaming复用组件,复用代码,有同一套行为。
  2. 如果想要Table/SQL的toDataset或者fromDataset,那就完全没戏了。尽可能的在Table的层面来处理吧。
  3. 后续我们正在考虑在DataStream上构建BoundedStream,给DataStream带来批处理的功能。

网络模型

图片 2.png

Batch模式就是在中间结果落盘,这个模式和典型的Batch处理是一致的,比如MapReduce/Spark/Tez。

Flink以前的网络模型也分为Batch和Pipeline两种,但是Batch模式只是支持上下游隔断执行,也就是说资源用量可以不用同时满足上下游共同的并发。但是另外一个关键点是Failover没有对接好,1.9和1.10在这方面进行了改进,支持了单点的Failover。

建议在Batch时打开:

jobmanager.execution.failover-strategy = region

为了避免重启过于频繁导致JobMaster太忙了,可以把重启间隔提高:

restart-strategy.fixed-delay.delay = 30 s

Batch模式的好处有:

  • 容错好,可以单点恢复
  • 调度好,不管多少资源都可以运行
  • 性能差,中间数据需要落盘,强烈建议开启压缩
    taskmanager.network.blocking-shuffle.compression.enabled = true

Batch模式比较稳,适合传统Batch作业,大作业。

图片 3.png

Pipeline模式是Flink的传统模式,它完全和Streaming作业用的是同一套代码,其实社区里Impala和Presto也是类似的模式,纯走网络,需要处理反压,不落盘,它主要的优缺点是:

  • 容错差,只能全局重来
  • 调度差,你得保证有足够的资源
  • 性能好,Pipeline执行,完全复用Stream,复用流控反压等功能。

有条件可以考虑开启Pipeline模式。

调度模型

Flink on Yarn支持两种模式,Session模式和Per job模式,现在已经在调度层次高度统一了。

  1. Session模式没有最大进程限制,当有Job需要资源时,它就会去Yarn申请新资源,当Session有空闲资源时,它就会给Job复用,所以它的模型和PerJob是基本一样的。
  2. 唯一的不同只是:Session模式可以跨作业复用进程。

另外,如果想要更好的复用进程,可以考虑加大TaskManager的超时释放:
resourcemanager.taskmanager-timeout = 900000

资源模型

先说说并发:

  1. 对Source来说:目前Hive的table是根据InputSplit来定需要多少并发的,它之后能Chain起来的Operators自然都是和source相同的并发。
  2. 对下游网络传输过后的Operators(Tasks)来说:除了一定需要单并发的Task来说,其它Task全部统一并发,由table.exec.resource.default-parallelism统一控制。

我们在Blink内部实现了基于统计信息来推断并发的功能,但是其实以上的策略在大部分场景就够用了。

Manage内存

图片 4.png

目前一个TaskManager里面含有多个Slot,在Batch作业中,一个Slot里只能运行一个Task (关闭SlotShare)。

对内存来说,单个TM会把Manage内存切分成Slot粒度,如果1个TM中有n个Slot,也就是Task能拿到1/n的manage内存。

我们在1.10做了重大的一个改进就是:Task中chain起来的各个operators按照比例来瓜分内存,所以现在配置的算子内存都是一个比例值,实际拿到的还要根据Slot的内存来瓜分。

这样做的一个重要好处是:

  1. 不管当前Slot有多少内存,作业能都run起来,这大大提高了开箱即用。
  2. 不管当前Slot有多少内存,Operators都会把内存瓜分干净,不会存在浪费的可能。

当然,为了运行的效率,我们一般建议单个Slot的manage内存应该大于500MB。

另一个事情,在1.10后,我们去除了OnHeap的manage内存,所以只有off-heap的manage内存。

外部系统集成

Hive

强烈推荐Hive Catalog + Hive,这也是目前批处理最成熟的架构。在1.10中,除了对以前功能的完善以外,其它做了几件事:

  1. 多版本支持,支持Hive 1.X 2.X 3.X
  2. 完善了分区的支持,包括分区读,动态/静态分区写,分区统计信息的支持。
  3. 集成Hive内置函数,可以通过以下方式来load:
    a)TableEnvironment.loadModule("hiveModule",new HiveModule("hiveVersion"))
  4. 优化了ORC的性能读,使用向量化的读取方式,但是目前只支持Hive 2+版本,且要求列没有复杂类型。有没有进行过优化差距在5倍量级。

兼容Streaming Connectors

得益于流批统一的架构,目前的流Connectors也能在batch上使用,比如HBase的Lookup和Sink、JDBC的Lookup和Sink、Elasticsearch的Sink,都可以在Batch无缝对接使用起来。

实践

SQL-CLI

在1.10中,SQL-CLI也做了大量的改动,比如把SQL-CLI做了stateful,里面也支持了DDL,还支持了大量的DDL命令,给SQL-CLI暴露了很多TableEnvironment的能力,这让用户可以方便得多。后续,我们也需要对接JDBC的客户端,让用户可以更好的对接外部工具。但是SQL-CLI仍然待继续改进,比如目前仍然只支持Session模式,不支持Per Job模式。

编程方式

TableEnvironment tEnv = TableEnvironment.create(EnvironmentSettings
  .newInstance()
  .useBlinkPlanner()
  .inBatchMode()
  .build());

老的BatchTableEnv因为绑定了Dataset,而且区分Java和Scala,是不干净的设计方式,所以Blink planner只支持新的TableEnv。

TableEnv注册的source, sink, connector, functions,都是temporary的,重启之后即失效了。如果需要持久化的object,考虑使用HiveCatalog。

tEnv.registerCatalog(“hive”, hiveCatalog);
tEnv.useCatalog(“hive”);

可以通过tEnv.sqlQuery来执行DML,这样可以获得一个Table,我们也通过collect来获得小量的数据:

Table table = tEnv.sqlQuery(“SELECT COUNT(*) FROM MyTable”);
List<Row> results = TableUtils.collectToList(table);
System.out.println(results);

可以通过tEnv.sqlUpdate来执行DDL,但是目前并不支持创建hive的table,只能创建Flink类型的table:

tEnv.sqlUpdate(
   "CREATE TABLE myResult (" +
      "  cnt BIGINT"
      ") WITH (" +
      "  'connector.type'='jdbc'," 
         ……
      ")");

可以通过tEnv.sqlUpdate来执行insert语句,Insert到临时表或者Catalog表中,比如insert到上面创建的临时JDBC表中:

tEnv.sqlUpdate(“INSERT INTO myResult SELECT COUNT(*) FROM MyTable”);
tEnv.execute(“MyJob”);

当结果表是Hive表时,可以使用Overwrite语法,也可以使用静态Partition的语法,这需要打开Hive的方言:

tEnv.getConfig().setSqlDialect(SqlDialect.HIVE);

结语

目前Flink batch SQL仍然在高速发展中,但是1.10已经是一个可用的版本了,它在功能上、性能上都有很大的提升,后续还有很多有意思的features,等待着大家一起去挖掘。

相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程,适合初学者阅读。由浅入深案例丰富,通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员,只要能够坚持把所有章节都学完,也一定会受益匪浅。
相关文章
|
6天前
|
SQL 存储 API
Flink(十五)【Flink SQL Connector、savepoint、CateLog、Table API】(5)
Flink(十五)【Flink SQL Connector、savepoint、CateLog、Table API】
|
6天前
|
SQL 消息中间件 Java
Flink(十五)【Flink SQL Connector、savepoint、CateLog、Table API】(4)
Flink(十五)【Flink SQL Connector、savepoint、CateLog、Table API】
|
6天前
|
SQL Java API
Flink(十五)【Flink SQL Connector、savepoint、CateLog、Table API】(3)
Flink(十五)【Flink SQL Connector、savepoint、CateLog、Table API】
|
6天前
|
SQL 关系型数据库 数据库
Flink(十五)【Flink SQL Connector、savepoint、CateLog、Table API】(2)
Flink(十五)【Flink SQL Connector、savepoint、CateLog、Table API】
|
3天前
|
SQL DataWorks NoSQL
DataWorks产品使用合集之如何将SQL Server中的数据转存到MongoDB
DataWorks作为一站式的数据开发与治理平台,提供了从数据采集、清洗、开发、调度、服务化、质量监控到安全管理的全套解决方案,帮助企业构建高效、规范、安全的大数据处理体系。以下是对DataWorks产品使用合集的概述,涵盖数据处理的各个环节。
|
27天前
|
SQL API 流计算
实时计算 Flink版产品使用合集之在Mac M1下的Docker环境中开启SQL Server代理的操作步骤是什么
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
139 1
|
20天前
|
SQL 存储 搜索推荐
SQL server增删改查(1)
SQL server增删改查(1)
75 0
|
21天前
|
SQL 关系型数据库 数据库
阿里云数据库 RDS SQL Server版实战【性能优化实践、优点探析】
本文探讨了Amazon RDS SQL Server版在云数据库中的优势,包括高可用性、可扩展性、管理便捷、安全性和成本效益。通过多可用区部署和自动备份,RDS确保数据安全和持久性,并支持自动扩展以适应流量波动。可视化管理界面简化了监控和操作,而数据加密和访问控制等功能保障了安全性。此外,弹性计费模式降低了运维成本。实战应用显示,RDS SQL Server版能有效助力企业在促销高峰期稳定系统并保障数据安全。阿里云的RDS SQL Server版还提供了弹性伸缩、自动备份恢复、安全性和高可用性功能,进一步优化性能和成本控制,并与AWS生态系统无缝集成,支持多种开发语言和框架。
147 2
|
21天前
|
SQL JSON atlas
实时计算 Flink版产品使用合集之SQL Server CDC是否支持抽取SQL Server视图
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
26天前
|
SQL 数据处理 API
实时计算 Flink版产品使用合集之遇到SQL Server锁表问题如何解决
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
27 0

相关产品

  • 实时计算 Flink版