Apache Hudi使用是怎么样的

阿帕奇胡迪是如何工作的？我相信很多没有经验的人对此无能为力。为此，本文总结了问题产生的原因及解决方法。希望你能通过这篇文章解决这个问题。]

00-1010实时可分为数据的实时处理和实时即席分析，这就需要对数据进行实时处理，并应立即获得相应的结果。Flink和Spark Streaming用于实时数据的实时处理，需要实时处理和快速处理。数据不实时、处理不及时的场景就是我们的仓库T 1数据。

本文讨论的阿帕奇胡迪，对应的场景是实时数据，而不是实时处理。它旨在将Mysql中的时间近乎实时地映射到Hive等大数据平台。

数据实时处理和实时的数据

传统的离线数据仓库，通常数据为T ^ 1，不能满足当天的数据分析需求，而流式计算一般基于窗口，且窗口逻辑相对固定。作者工作的公司有一种特殊的需求。业务分析熟悉现有交易数据库的数据结构，希望会有大量的即席分析，包括当天的实时数据。他们曾经直接通过基于Mysql从库的Sql做相应的分析计算。但是很多时候你会遇到以下障碍。

当数据量大，分析逻辑复杂时，Mysql从数据库下载需要很长时间。

一些跨库分析无法实现。

因此，出现了一些弥合OLTP和OLAP之间差距的技术框架，通常是TiDB。它可以支持OLTP和OLAP。而Apache胡迪和Apache Kudu则相当于OLTP和OLAP技术之间的现有桥梁。它们可以将数据存储在现有的OLTP数据结构中，支持CRUD，并提供与现有OLAP框架(如Hive、Impala)的集成，实现OLAP分析。

Apache Kudu，需要单独部署集群。阿帕奇胡迪不需要它。它可以利用现有的大数据集群如HDFS存储数据文件，然后利用Hive做数据分析。相对来说更适合资源受限的环境# #阿帕奇胡迪入门。

业务场景和技术选型

胡迪提供了支持CRUD操作的胡迪表的概念。基于这个特性，我们可以将Mysql Binlog数据重放到胡迪表，然后基于Hive对胡迪表进行查询和分析。数据流结构如下

使用Aapche Hudi整体思路

胡迪数据文件可以存储在操作系统的文件系统或分布式文件系统HDFS中。HDFS通常用于存储，以便随后分析性能和数据可靠性。就HDFS存储而言，胡迪表的存储文件可以分为两类。

与_partition_key相关的路径是按分区存储的实际数据文件。当然可以指定分区的路径键。我在这里使用了_partition_key。帽衫由于CRUD的分散性，每个操作都会生成一个文件。当小文件越来越多的时候，会严重影响HDFS的表现。胡迪设计了一种文件合并机制。那个。hoodie文件夹存储与相应文件合并操作相关的日志文件。

Hudi表数据结构

胡迪真实数据文件以拼花文件格式存储

数据文件

随着时间的推移，胡迪在时间线上调用了一系列CRUD操作。时间轴中的操作称为即时。即时包含以下信息

实例化记录这个操作是数据提交、文件合并还是文件清理。

瞬间时间此操作发生的时间。

状态操作的状态，已启动(已请求)、正在进行(正在进行)或已完成(已完成)

那个。帽衫文件夹存储相应操作的状态记录

.hoodie文件

胡迪需要能够唯一地识别记录，以便实现数据的CRUD。胡迪会的

把数据集中的唯一字段(record key ) + 数据所在分区 (partitionPath) 联合起来当做数据的唯一键

COW和MOR

基于上述基础概念之上，Hudi提供了两类表格式COW和MOR。他们会在数据的写入和查询性能上有一些不同

Copy On Write Table

简称COW。顾名思义，他是在数据写入的时候，复制一份原来的拷贝，在其基础上添加新数据。正在读数据的请求，读取的是是近的完整副本，这类似Mysql 的MVCC的思想。

上图中，每一个颜色都包含了截至到其所在时间的所有数据。老的数据副本在超过一定的个数限制后，将被删除。这种类型的表，没有compact instant，因为写入时相当于已经compact了。

• 优点 读取时，只读取对应分区的一个数据文件即可，较为高效

• 缺点 数据写入的时候，需要复制一个先前的副本再在其基础上生成新的数据文件，这个过程比较耗时。且由于耗时，读请求读取到的数据相对就会滞后

Merge On Read Table

简称MOR。新插入的数据存储在delta log 中。定期再将delta log合并进行parquet数据文件。读取数据时，会将delta log跟老的数据文件做merge，得到完整的数据返回。当然，MOR表也可以像COW表一样，忽略delta log，只读取最近的完整数据文件。下图演示了MOR的两种数据读写方式

• 优点 由于写入数据先写delta log，且delta log较小，所以写入成本较低

• 缺点 需要定期合并整理compact，否则碎片文件较多。读取性能较差，因为需要将delta log 和 老数据文件合并

基于hudi的代码实现

我在github上放置了基于Hudi的封装实现，对应的源码地址为 https://github.com/wanqiufeng/hudi-learn。

binlog数据写入Hudi表

• binlog-consumer分支使用Spark streaming消费kafka中的Binlog数据，并写入Hudi表。Kafka中的binlog是通过阿里的Canal工具同步拉取的。程序入口是CanalKafkaImport2Hudi，它提供了一系列参数，配置程序的执行行为

一个使用的demo如下

/data/opt/spark-2.4.4-bin-hadoop2.6/bin/spark-submit --class com.niceshot.hudi.CanalKafkaImport2Hudi \
	--name hudi__goods \
    --master yarn \
    --deploy-mode cluster \
    --driver-memory 512m \
    --executor-memory 512m \
    --executor-cores 1 \
	--num-executors 1 \
    --queue hudi \
    --conf spark.executor.memoryOverhead=2048 \
    --conf "spark.executor.extraJavaOptions=-XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=\tmp\hudi-debug" \
	--conf spark.core.connection.ack.wait.timeout=300 \
	--conf spark.locality.wait=100 \
	--conf spark.streaming.backpressure.enabled=true \
	--conf spark.streaming.receiver.maxRate=500 \
	--conf spark.streaming.kafka.maxRatePerPartition=200 \
	--conf spark.ui.retainedJobs=10 \
	--conf spark.ui.retainedStages=10 \
	--conf spark.ui.retainedTasks=10 \
	--conf spark.worker.ui.retainedExecutors=10 \
	--conf spark.worker.ui.retainedDrivers=10 \
	--conf spark.sql.ui.retainedExecutions=10 \
	--conf spark.yarn.submit.waitAppCompletion=false \
	--conf spark.yarn.maxAppAttempts=4 \
	--conf spark.yarn.am.attemptFailuresValidityInterval=1h \
	--conf spark.yarn.max.executor.failures=20 \
	--conf spark.yarn.executor.failuresValidityInterval=1h \
	--conf spark.task.maxFailures=8 \
    /data/opt/spark-applications/hudi_canal_consumer/hudi-canal-import-1.0-SNAPSHOT-jar-with-dependencies.jar  --kafka-server local:9092 --kafka-topic dt_streaming_canal_xxx --base-save-path hdfs://192.168.2.1:8020/hudi_table/ --mapping-mysql-db-name crm --mapping-mysql-table-name order --primary-key id --partition-key createDate --duration-seconds 1200

历史数据同步以及表元数据同步至hive

history_import_and_meta_sync 分支提供了将历史数据同步至hudi表，以及将hudi表数据结构同步至hive meta的操作

同步历史数据至hudi表

这里采用的思路是

• 将mysql全量数据通过注入sqoop等工具，导入到hive表。

• 然后采用分支代码中的工具HiveImport2HudiConfig，将数据导入Hudi表

HiveImport2HudiConfig提供了如下一些参数，用于配置程序执行行为

一个程序执行demo

nohup java -jar hudi-learn-1.0-SNAPSHOT.jar --sync-hive-db-name hudi_temp --sync-hive-table-name crm__wx_user_info --base-save-path hdfs://192.168.2.2:8020/hudi_table/ --mapping-mysql-db-name crm --mapping-mysql-table-name "order" --primary-key "id" --partition-key created_date --hive-site-path /etc/lib/hive/conf/hive-site.xml --tmp-data-path /data/tmp > order.log &

同步hudi表结构至hive meta

需要将hudi的数据结构和分区，以hive外表的形式同步至Hive meta，才能是Hive感知到hudi数据，并通过sql进行查询分析。Hudi本身在消费Binlog进行存储时，可以顺带将相关表元数据信息同步至hive。但考虑到每条写入Apache Hudi表的数据，都要读写Hive Meta ，对Hive的性能可能影响很大。所以我单独开发了HiveMetaSyncConfig工具，用于同步hudi表元数据至Hive。考虑到目前程序只支持按天分区，所以同步工具可以一天执行一次即可。参数配置如下

一个程序执行demo

java -jar hudi-learn-1.0-SNAPSHOT.jar --hive-db-name streaming --hive-table-name crm__order --hive-user-name hive --hive-pwd hive --hive-jdbc-url jdbc:hive2://192.168.16.181:10000 --hudi-table-path hdfs://192.168.16.181:8020/hudi_table/crm__order --hive-site-path /lib/hive/conf/hive-site.xml

一些踩坑

hive相关配置

有些hive集群的hive.input.format配置，默认是org.apache.hadoop.hive.ql.io.CombineHiveInputFormat，这会导致挂载Hudi数据的Hive外表读取到所有Hudi的Parquet数据，从而导致最终的读取结果重复。需要将hive的format改为org.apache.hadoop.hive.ql.io.HiveInputFormat，为了避免在整个集群层面上更改对其余离线Hive Sql造成不必要的影响，建议只对当前hive session设置set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;

spark streaming的一些调优

由于binlog写入Hudi表的是基于Spark streaming实现的，这里给出了一些spark 和spark streaming层面的配置，它能使整个程序工作更稳定

未来改进

• 支持无分区，或非日期分区表。目前只支持日期分区表

• 多数据类型支持，目前为了程序的稳定性，会将Mysql中的字段全部以String类型存储至Hudi

看完上述内容，你们掌握Apache Hudi使用是怎么样的的方法了吗？如果还想学到更多技能或想了解更多相关内容，欢迎关注行业资讯频道，感谢各位的阅读！