基于TIS构建Apache Hudi千表入湖方案

ApacheHudi

发布于 2022-07-11 13:52:02

1.6K0

发布于 2022-07-11 13:52:02

拥抱数据湖

随着大数据时代的到来，数据量动辄PB级，因此亟需一种低成本、高稳定性的实时数仓解决方案来支持海量数据的OLAP查询需求，Apache Hudi[1]应运而生。Hudi借助与存放在廉价的分布式文件系统之中列式存储文件，并将其元数据信息存放在Hive元数据库中与传统查询引擎Hive、Presto、Spark等整合，完美地实现了计算与存储的分离。Hudi数据湖方案比传统的Hive数仓的优势是加入了数据实时同步功能，可以通过最新的Flink流计算引擎来以最小的成实现数据实时同步。本质来说Hudi是整合现有的技术方案实现的，属于新瓶装旧酒，Hudi内部需要整合各种组件（存储、Indexer、Compaction，文件分区），为了达到通用及灵活性，每个组件会有大量的配置参数需要设置，且各种组件的配置是有关联性的，所以对与新手来说要构建一个生产环境中可用的数据库方案，面对一大堆配置往往会望而却步。本文就向大家介绍如何通过TIS来改善Hudi数据湖实例构建流程，从而大幅提高工作效率。

TIS可以为您做什么？

TIS将Hudi中的各组件进行优雅地封装，并且基于TIS的数据字典组件自动生成Hudi DeltaStreamer[2]及 Flink Stream API[3]运行所需要配置，Hudi数据表相关的配置都是在TIS的UI界面上完成操作，实现了轻点鼠标完成构建流程，实现开箱即用从而大幅提高构建Hudi数据湖的效率。

TIS采用两种方式实现数据入湖：

1. DeltaStreamer: 该方法实现批量数据导入，通过DataX将数据表中数据以avro格式导入到HDFS中，之后启动DeltaStreamer通过Spark RDD消费HDFS中的原始数据进行数据入湖。该种方式适合历史数据导入，优点是速度快吞吐率大。不足是无法提供Hudi表增量同步功能。
2. 基于Flink Stream API[4]方式（Stream SQL不推荐，原因是，使用Stream API的方式可以和DeltaStreamer执行流程中都依赖同一份Avro Schema来定义Hudi表结构，这样就保证Hudi表数据结构统一）：基于Flink Stream API的方式来实现增量数据同步功能，优点是可以保证数据源和Hudi表保证低延时同步（一个CheckPoint周期之内），缺点是当利用该种方式结合Flink CDC组件来导入历史全量数据时由于触发CheckPoint执行过程种需要将历史数据写入到Flink Statebackend种存储，由于数据量大往往会导致Flink CheckPoint执行超时，导致Flink Job执行失败，另外，由于Flink Job执行过程种还会触发Hudi Compaction操作由于数据量大也会导致Flink Job产生OOM异常从而任务失败。

推荐用户在实际生产环境中将以上两种方式结合使用，初次构建Hudi，数据湖表需要导入历史全量数据，这时采用第一种DeltaStreamer批量导入历史全量数据。等待构建完成，继续开启Flink增量同步任务（以Flink CDC MySQL Connector[5]为例，消费游标使用Latest策略消费，消费最新Binlog增量数据）这样就可以规避Flink Job由于读取全量历史数据因数据量大导致的Job失败的故障产生。

实现原理

实现原理如下图所示：

通过在TIS平台中定义Hudi数据湖实例，可以选择由DeltaStreamer来执行数据批量导入Hudi表，或者由Flink Job来执行增量Hudi表数据同步任务。

如何安装

TIS的安装非常方便，只需三个步骤：下载一个tar压缩包，解压，最后启动即可。详细请查看 TIS安装说明[6]](http://tis.pub/docs/install/uber/))

TIS是基于微内核架构来实现的，初始安装包只有200兆，其他具体执行逻辑相关的执行逻辑都封装在了TIS的插件系统中，

本文相关的Hudi和MySQL以及Flink增量同步、DataX批量同步相关的功能插件都是在TIS运行时按需加载并热部署生效的。

依赖版本

本示例依赖如下组件版本：

组件名称	版本
Apache Hudi	0.10.1
Apache Spark	spark-2.4.4-bin-hadoop2.7
Apache Hive	2.1.1 以上
Apache Hadoop	2.7.3
Apache Flink	tis-1.13.1(基于Flink 1.13.1 定制，解决不同组件Source，Sink之间可能存在的三方依赖包冲突)

创建MySQL到Hudi千表入湖通道

准备工作

1. 准备一个MySQL测试数据库，版本5.7
2. 下载Hudi Source包[7]运行环境（使用Hudi0.10.1）：为了简化演示，在Hudi安装包中一个一个Hudi docker compose[8]启动脚本，本说明中就用该执行脚本来创建Hudi运行环境，详细请查看https://hudi.apache.org/docs/next/docker_demo完成Hudi Docker运行环境安装，启动docker-compose之前需要修改docker-compose的配置文件（./hudi-release-0.10.1/docker/compose/docker-compose_hadoop284_hive233_spark244.yml），在docker镜像实例sparkmaster与spark-worker-1上添加一个hosts配置文件可以避免DeltatStreamer执行过程中提交任务端Hostname不能识别的错误：

添加项：

extra_hosts:
  - "baisui-test-1:192.168.28.200"

添加后效果：

      sparkmaster:
           image: apachehudi/hudi-hadoop_2.8.4-hive_2.3.3-sparkmaster_2.4.4:latest
           hostname: sparkmaster
           container_name: sparkmaster
           env_file:
             - ./hadoop.env
           ports:
             - "8080:8080"
             - "7077:7077"
           extra_hosts:
             - "baisui-test-1:192.168.28.200"
           environment:
             - INIT_DAEMON_STEP=setup_spark
           links:
             - "hivemetastore"
             - "hiveserver"
             - "hive-metastore-postgresql"
             - "namenode"
       
         spark-worker-1:
           image: apachehudi/hudi-hadoop_2.8.4-hive_2.3.3-sparkworker_2.4.4:latest
           hostname: spark-worker-1
           container_name: spark-worker-1
           env_file:
             - ./hadoop.env
           depends_on:
             - sparkmaster
           ports:
             - "8081:8081"
           extra_hosts:
             - "baisui-test-1:192.168.28.200"
           environment:
             - "SPARK_MASTER=spark://sparkmaster:7077"
           links:
             - "hivemetastore"
             - "hiveserver"
             - "hive-metastore-postgresql"
             - "namenode"

然后就可以启动Hudi docker-compose了

# 启动 
sh hudi-0.10.1/docker/setup_demo.sh
# 停止
sh hudi-0.10.1/docker/stop_demo.sh

1. 安装tis-flink[9]](http://tis.pub/docs/install/flink-cluster/standalone/)) TIS整合了各种Source和Sink的插件，他们都是以插件(tpi)的方式封装的，为了实现开箱即用、避免插件之间三方包冲突，插件与插件之间是需要进行ClassLoader方式隔离，为此TIS对Flink13.1进行扩展（运行时TIS必须搭配TIS定制Flink，否则无法正常使用）
2. 安装单机版tis[10]

基本信息配置

1. 当完成安装步骤之后，进入TIS操作界面，点击菜单栏中实例链接

2. 进入实例列表，点击右侧添加下拉按钮中的数据管道，进行MySQL端到Hudi端的数据同步通道构建

3. 添加流程一共分为5步，第1步添加数据通道的基本信息

4. 进入数据端选择步骤，选择Reader Writer类型选择，由于系统刚安装，数据端类型对应的插件还没有选取，需要点击插件安装添加按钮，安装插件

5. 从插件列表中选择tis-ds-mysql-plugin,tis-datax-hudi-plugin两个插件进行安装

6. 插件安装完毕，将插件管理页面关闭

7. Reader端选择MySQL,Writer端选择Hudi，点击下一步按钮，进行MySQL Reader的设置

8. 在Reader设置页面，点击数据库名项右侧配置下拉框中MySqlV5 数据源，完成表单填写，点击保存按钮，其他输入项目使用默认值即可，然后再点击下一步选取Reader端中需要处理的表

9. 选择需要的表: 点击设置按钮，对目标Hudi表设置，设置目标表的目标列、分区、主键等属性设置.

点击保存按钮，然后点击下一步，进入Hudi Writer表单设置

Hudi Writer表单

1. 点击sparkConn项右侧管理下拉框中添加按钮，添加SparkMaster源

2. 点击hiveConn项右侧 数据源管理下拉框添加按钮，添加hiveConn源

3. 点击fsName项右侧 FS管理 下拉框添加按钮，添加分布式文件系统源

4. 其他选项按照说明设置录入确认页面，对上几步流程中录入的配置信息进行确认

5. 点击创建按钮完成数据流通道定义

批量数据导入

1. Hudi数据通道定义完成，开始批量导入MySQL中全量历史数据

2. 由于TIS系统初次安装，还未安装触发器插件，请按照只是先安装，成功之后再触发批量导入

Flink增量同步开通

接下来开通实时增量通道

1. 首先需要安装Flink单机版 安装说明[11]](http://tis.pub/docs/install/flink-cluster/standalone/))

2. Flink集群启动之后，在TIS中添加Flink集群对应配置,请妥善设置Flink相关的重启策略、checkpoint、stateBackend相关属性

3. 表单填写完成之后，点击保存&下一步按钮进入下一步Sink，Source相关属性设置该步骤添加Flink SourceFunction对应的flink-connector-mysql-cdc[12]](https://github.com/ververica/flink-cdc-connectors/tree/master/flink-connector-mysql-cdc))插件和 Fink Sink对应的Hudi 插件

4. 设置完成之后进入下一步

5. TIS会解析Reader选取的表元数据信息，自动生成Flink Stream Code 在该版本中，自动生成的Flink Stream Code还不支持用户自定义编写业务逻辑

6. 点击部署按钮，进入向Flink Cluster中部署流处理逻辑等待片刻，跳转到创建成功的页面

7. 至此，MySQL与Hudi表增量通道已经添加完成，MySQL到Hudi表实时数据同步可以保证在一个Checkpoint周期内完成，接下来可以尝试在MySQL数据表上，更新几条数据，然后在Hudi 对应的表上验证更新是否成功同步。

总结

通过以上流程介绍，我们发现通过使用TIS[13]](http://tis.pub))来实现MySQL与Hudi表同步有如下优势：

• 安装方便，组件按需加载，热生效
• 支持数据源分库，多表同步
• 完美实现低代码配置DataOps的目标，帮助用户大大提高工作效率，且避免出错

还等什么呢？赶快试用一下吧

后续

本次是TIS与数据湖产品Hudi的整合的初次尝试，Hudi的配置项比较繁杂，且各个配置项之间又存在各种依赖关系。TIS对Hudi的封装过程中还没有将Hudi的所有配置项（例如：Indexer的配置相关）开放给用户配置，后续会陆续完善。假如您TIS有功能需求，请在Github的Issue栏中添加：https://github.com/qlangtech/tis/issues[14]

引用链接

[1] Apache Hudi: [https://hudi.apache.org/ [2] Hudi DeltaStreamer: https://hudi.apache.org/docs/next/hoodie_deltastreamer#deltastreamer [3] Flink Stream API: https://github.com/apache/hudi/blob/master/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/streamer/HoodieFlinkStreamer.java [4] Flink Stream API: https://github.com/apache/hudi/blob/master/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/streamer/HoodieFlinkStreamer.java [5] Flink CDC MySQL Connector: https://github.com/ververica/flink-cdc-connectors/tree/master/flink-connector-mysql-cdc [6] TIS安装说明: [http://tis.pub/docs/install/uber/ [7] Source包: https://www.apache.org/dyn/closer.lua/hudi/0.10.1/hudi-0.10.1.src.tgz [8] docker compose: https://www.cnblogs.com/yang-qiu/p/15726821.html [9] tis-flink: [http://tis.pub/docs/install/flink-cluster/standalone/ [10] 安装单机版tis: http://tis.pub/docs/install/uber/ [11] 安装说明: [http://tis.pub/docs/install/flink-cluster/standalone/ [12] flink-connector-mysql-cdc: [https://github.com/ververica/flink-cdc-connectors/tree/master/flink-connector-mysql-cdc [13] TIS: [http://tis.pub [14] https://github.com/qlangtech/tis/issues: https://github.com/qlangtech/tis/issues](https://github.com/qlangtech/tis/issues)