Flink1.10和Hive3.0性能对比

如今的大数据批计算&＃xff0c;随着 Hive 数仓的成熟&＃xff0c;普遍的模式是 Hive metastore &＃43; 计算引擎。常见的计算引擎有 Hive on MapReduce、Hive on Tez、Hive on Spark、Spark integrate Hive、Presto integrate Hive&＃xff0c;还有随着 Flink 1.10 发布后生产可用的 Flink Batch SQL。

Flink 作为一个统一的计算引擎&＃xff0c;旨在提供统一的流批体验以及技术栈。Flink 在 1.9 合并了 Blink 的代码&＃xff0c;并在 1.10 中完善了大量的功能以及性能&＃xff0c;可以运行所有 TPC-DS 的查询&＃xff0c;性能方面也很有竞争力&＃xff0c;Flink 1.10 是一个生产可用的、批流统一的 SQL 引擎版本。

在搭建计算平台的过程中&＃xff0c;性能和成本是选取计算引擎的很关键的因素。为此&＃xff0c;Ververica 的 flink-sql-benchmark [1] 项目提供了基于 Hive Metastore 的 TPC-DS Benchmark 测试的工具&＃xff0c;为了测试更靠近真正的生产作业&＃xff1a;

• 测试的输入表都是标准的 Hive 表&＃xff0c;数据全在与生产一致的 Hive 数仓中。其它计算引擎也能方便分析这些表。
• 数据的格式采用 ORC&＃xff0c;ORC 是常用的生产文件格式&＃xff0c;提供较高的压缩率&＃xff0c;和较好的读取性能。
• 选取 TPC-DS Benchmark 的 10TB 数据集&＃xff0c;10TB 的数据集是比较常见的生产规模。如果只有 1TB&＃xff0c;完全可以在传统数据库中运行起来&＃xff0c;不太适合大数据的测试。

我们在 20 台机器上测试了三种引擎&＃xff1a;Flink 1.10、Hive 3.0 on MapReduce、Hive 3.0 on Tez&＃xff0c;从两个维度测试了引擎的成绩&＃xff1a;

• 总时长&＃xff1a;直观的性能数据&＃xff0c;但是可能会受到个别 queries 的较大影响。
• 几何平均&＃xff1a;表示一组数的中心趋势&＃xff0c;它可以更好的消除个别 queries 的较大影响&＃xff0c;呈现较真实的平均数。

结果摘要&＃xff1a;

• Flink 1.10 VS Hive 3.0 on MapReduce

  • Flink 总时长的性能是 Hive on MapReduce 的 8.7 倍。
  • Flink Queries 几何平均的性能是 Hive on MapReduce 的 7.8 倍。

• Flink 1.10 VS Hive 3.0 on Tez

  • Flink 总时长的性能是 Hive on Tez 的 2.1 倍。
  • Flink Queries 几何平均的性能是 Hive on Tez 的 2.0 倍。

运行总时间的对比成绩是&＃xff1a;

Queries 几何平均的对比成绩是&＃xff1a;

本文只测试了上述引擎和 10TB 的数据集&＃xff0c;读者可以根据自己的集群规模&＃xff0c;选取特定的数据集&＃xff0c;使用 flink-sql-benchmark 工具来运行更多引擎的对比测试。

Benchmark 详情

Benchmark 环境

具体环境及调优说明&＃xff1a;

• 计算环境&＃xff1a;20 台机器&＃xff0c;机器参数为 64 核 intel 处理器、256GB 内存、1 SSD 盘用于计算引擎、多块 SATA 盘用于 HDFS、万兆网卡。
• 集群环境&＃xff1a;Yarn &＃43; HDFS &＃43; Hive。
• Flink参数&＃xff1a;flink-conf.yaml [2]。
• Hive参数&＃xff1a;主要调优了 MapJoin 的阈值&＃xff0c;提高性能的同时避免 OOM。
• 选用较新的 Hadoop 版本(3.X)&＃xff0c;并选用了较新的 Hive 和 Tez 版本

Benchmark 步骤

■环境准备

1. 准备 Hadoop (HDFS &＃43; YARN) 环境
2. 准备 Hive 环境

■ 数据集生成

1. 分布式生成 TPC-DS 数据集&＃xff0c;并加载 TEXT 数据集到 Hive&＃xff0c;原始数据是 Csv 的格式。建议分布式生成数据&＃xff0c;这也是个比较耗时的步骤。&＃xff08;flink-sql-benmark 工具中集成了 TPC-DS 的工具&＃xff09;
2. Hive TEXT 表转换为 ORC 表&＃xff0c;ORC 格式是常见的 Hive 数据文件格式&＃xff0c;行列混合的存储有利于后续的快速分析&＃xff0c;也有很高的压缩比。执行 Query&＃xff1a;create table ${NAME} stored as ${FILE} as select * from $SOURCE.${NAME};

如图&＃xff0c;生成了 TPC-DS 官方说明的 7 张事实表和 17 张维表。

1. 分析 Hive 表&＃xff0c;统计信息对于分析作业的查询优化非常重要&＃xff0c;对于复杂的 SQL&＃xff0c;Plan 的执行效率有很大的差异。Flink 不但支持读取 Hive 的 Table 统计信息&＃xff0c;也支持读取 Hive 的分区统计信息&＃xff0c;根据统计信息进行 CBO 的优化。执行命令&＃xff1a;analyze table ${NAME} compute statistics for columns;

■Flink 运行 Queries

1. 准备 Flink 环境&＃xff0c;搭建 Flink Yarn Session 环境&＃xff0c;推荐使用 Standalone 或者 Session 模式&＃xff0c;可以复用 Flink 的进程&＃xff0c;加快分析型作业的速度。
2. 编写代码运行 Queries&＃xff0c;统计执行时间等相关信息&＃xff0c;具体代码可以直接复用 flink-sql-benchmark 里的 flink-tpcds 工程。
3. FLINK_HOME/flink run 运行程序&＃xff0c;执行所有 queries&＃xff0c;等待执行完毕&＃xff0c;统计执行时间。

■其它引擎运行 Queries

1. 根据其它引擎的官网提示&＃xff0c;搭建环境。
2. 得益于标准的 Hive 数据集&＃xff0c;可以方便的使用其它引擎来读取 Hive 数据。
3. 在运行时&＃xff0c;值得注意的是需要达到集群的瓶颈&＃xff0c;比如 Cpu、比如 Disk&＃xff0c;一定是有瓶颈出现&＃xff0c;才能证明运行方式和参数是比较合理的&＃xff0c;为此&＃xff0c;需要一些性能调优。

Benchmark 分析

Flink 1.10

Flink 1.9 在合并 Blink 代码的时候&＃xff0c;就已经完成了很多工作&＃xff1a;深度 CodeGeneration、Binary 存储与计算、完善的 CBO 优化、Batch Shuffler&＃xff0c;为后续的性能突破打下了扎实的基础。

Flink 1.10 继续完善 Hive 集成&＃xff0c;并达到了生产级别的 Hive 集成标准&＃xff0c;其它也在性能和开箱即用方面做了很多工作&＃xff1a;

• Hive 多版本的支持&＃xff0c;支持了 Hive 1.0 以后的主要版本。

• 向量化的 ORC 读&＃xff0c;目前只在 Hive 2.0 以上版本才会默认开启。

  • Hive 1.X 版本的支持已经在进行中&＃xff1a;FLINK-14802 [3]
  • Parquet 的向量化读支持也已经在开发中&＃xff1a;FLINK-11899 [4]

• 基于比例的弹性内存分配&＃xff0c;这不仅利于 Operator 可以更多的使用内存&＃xff0c;而且大大方便了用户的配置&＃xff0c;用户不再需要配置 Operator 内存&＃xff0c;Operator 根据 Slot 弹性的拿到内存&＃xff0c;提高了 Flink 开箱即用的易用性。详见 FLIP-53 [5]

• Shuffle 的压缩&＃xff1a;Flink 默认给 Batch 作业开启中间数据落盘的方式&＃xff0c;这有利于避免调度死锁的可能&＃xff0c;也提供了良好的容错机制&＃xff0c;但是大量的落盘可能导致作业瓶颈在磁盘的吞吐上&＃xff0c;所以 Flink 1.10 开发了 Shuffle 的压缩&＃xff0c;用 Cpu 换 IO。

• 新调度框架&＃xff1a;Flink 1.10 也引入新了的调度框架&＃xff0c;这有利于 JobMaster 的调度性能&＃xff0c;避免并发太大时&＃xff0c;JobMaster 成为性能瓶颈。

Flink 参数分析

Flink 1.10 做了很多参数的优化&＃xff0c;提高用户的开箱即用体验&＃xff0c;但是由于批流一体的一些限制&＃xff0c;目前也是需要进行一些参数设置的&＃xff0c;这里本文粗略分析下。

■Table 层参数&＃xff1a;

• table.optimizer.join-reorder-enabled &＃61; true&＃xff1a;需要手动打开&＃xff0c;目前各大引擎的 JoinReorder 少有默认打开的&＃xff0c;在统计信息比较完善时&＃xff0c;是可以打开的&＃xff0c;一般来说 reorder 错误的情况是比较少见的。
• table.optimizer.join.broadcast-threshold &＃61; 10_1024_1024&＃xff1a;从默认值 1MB 调整到 10MB&＃xff0c;目前 Flink 的广播机制还有待提高&＃xff0c;所以默认值为 1MB&＃xff0c;但是在并发规模不是那么大的情况下&＃xff0c;可以开到 10MB。
• table.exec.resource.default-parallelism &＃61; 800&＃xff1a;Operator 的并发设置&＃xff0c;针对 10T 的输入&＃xff0c;建议开到 800 的并发&＃xff0c;不建议太大并发&＃xff0c;并发越大&＃xff0c;对系统各方面的压力越大。

■TaskManager 参数分析&＃xff1a;

• taskmanager.numberOfTaskSlots &＃61; 10&＃xff1a;单个 TM 里的 slot 个数。
• TaskManager 内存参数&＃xff1a;TaskManager 的内存主要分为三种&＃xff0c;管理内存、网络内存、JVM 相关的其它内存。需要理解下官网的文档&＃xff0c;才能有效的设置这些参数。
• taskmanager.memory.process.size &＃61; 15000m&＃xff1a;TaskManager 的总内存&＃xff0c;减去其它内存后一般留给堆内 3-5GB 的内存。
• taskmanager.memory.managed.size &＃61; 8000m&＃xff1a;管理内存&＃xff0c;用于 Operator 的计算&＃xff0c;留给单个 Slot 300 - 800MB 的内存是比较合理的。
• taskmanager.network.memory.max &＃61; 2200mb&＃xff1a;Task 点到点的通信需要 4 个 Buffers&＃xff0c;根据并发大概计算得出需要 2GB&＃xff0c;可以通过尝试得出&＃xff0c;Buffers 不够会抛出异常。

■网络参数分析

• taskmanager.network.blocking-shuffle.type &＃61; mmap&＃xff1a;Shuffle read 使用 mmap 的方式&＃xff0c;直接靠系统来管理内存&＃xff0c;是比较方便的形式。
• taskmanager.network.blocking-shuffle.compression.enabled &＃61; true&＃xff1a;Shuffle 使用压缩&＃xff0c;这个参数是批流复用的&＃xff0c;强烈建议给批作业开启压缩&＃xff0c;不然瓶颈就会在磁盘上。

■调度参数分析

• cluster.evenly-spread-out-slots &＃61; true&＃xff1a;在调度 Task 时均匀调度到每个 TaskManager 中&＃xff0c;这有利于使用所有资源。
• jobmanager.execution.failover-strategy &＃61; region&＃xff1a;默认全局重试&＃xff0c;需打开 region 重试才能 enable 单点的 failover。
• restart-strategy &＃61; fixed-delay&＃xff1a;重试策略需要手动设置&＃xff0c;默认是不重试的。

其它 timeout 相关参数是为了避免调度和运行过程中&＃xff0c;大数据量导致的网络抖动&＃xff0c;进而导致作业失败的问题。

Flink 1.11 及后续规划

后续 Flink 社区会在完善功能的同时进一步夯实性能&＃xff1a;

• 提供 SQL Gateway 以及 JDBC Driver&＃xff0c;目前提供独立仓库&＃xff0c;面向 Flink 1.10。[6] [7]
• 提供 Hive 语法兼容模式&＃xff0c;避免 Hive 用户的困扰。
• 完善 ORC 和 Parquet 的向量化读。
• N-Ary stream operator [8]&＃xff1a;开发 table 层的 chain 框架&＃xff0c;进一步避免 Shuffle 落盘导致的开销。

查看更多&＃xff1a;https://yqh.aliyun.com/detail/6264?utm_content&＃61;g_1000105248

上云就看云栖号&＃xff1a;更多云资讯&＃xff0c;上云案例&＃xff0c;最佳实践&＃xff0c;产品入门&＃xff0c;访问&＃xff1a;https://yqh.aliyun.com/