使用 Hudi 和 MinIO 开发流数据湖

Apache Hudi 是一个流数据湖平台，可将核心仓库和数据库功能直接引入数据湖。 Hudi 不满足于称自己为 或 等开放文件格式，它提供表、事务、更新插入/删除、高级索引、流式摄取服务、数据集群/压缩优化和并发性。 Delta Apache Iceberg Hudi 于 2016 年推出，牢牢扎根于 Hadoop 生态系统，其名称背后的含义是：Hadoop Upserts and Incrementals。它是为了管理 HDFS 上大型分析数据集的存储而开发的。 Hudi 的主要目的是减少流数据摄取期间的延迟。 随着时间的推移，Hudi 已经发展到使用 和对象存储，包括 MinIO。 Hudi 从 HDFS 的转变与世界的大趋势齐头并进，抛弃传统的 HDFS，转向高性能、可扩展和云原生对象存储。 Hudi 承诺提供优化，使 Apache Spark、Flink、Presto、Trino 等的分析工作负载更快，这与 MinIO 大规模云原生应用程序性能的承诺非常吻合。 云存储 在生产中使用 Hudi 的公司包括 、 、 和 。这些是世界上最大的一些 。在这个用例中，Hudi 的关键在于它提供了一个 数据处理堆栈，可以对柱状数据进行低延迟处理。通常，系统使用 Apache Parquet 或 ORC 等开放文件格式一次性写出数据，并将其存储在高度可扩展的对象存储或分布式文件系统之上。 Hudi 充当数据平面来摄取、转换和管理这些数据。 Hudi 使用 与存储进行交互，该 API 与从 HDFS 到对象存储再到内存文件系统的实现兼容（但不一定是最佳的）。 Uber Amazon ByteDance Robinhood 流数据湖 增量 Hadoop FileSystem API Hudi 文件格式 Hudi 使用基本文件和增量日志文件来存储对给定基本文件的更新/更改。基础文件可以是 Parquet（列式）或 HFile（索引式）。增量日志保存为 （行），因为在基本文件发生更改时记录这些更改是有意义的。 Avro Hudi 将给定基本文件的所有更改编码为块序列。块可以是数据块、删除块或回滚块。这些块被合并以获得更新的基础文件。此编码还会创建一个独立的日志。 。 来源 Hudi表格式 表格式由表的文件布局、表的架构以及跟踪表更改的元数据组成。 Hudi 强制执行写入模式，这与对流处理的强调一致，以确保管道不会因非向后兼容的更改而中断。 Hudi 将给定表/分区的文件分组在一起，并在记录键和文件组之间进行映射。如上所述，所有更新都记录到特定文件组的增量日志文件中。这种设计比 Hive ACID 更高效，Hive ACID 必须将所有数据记录与所有基础文件合并才能处理查询。 Hudi 的设计预期基于键的快速更新插入和删除，因为它适用于文件组的增量日志，而不是整个数据集。 Hudi 将给定表/分区的文件分组在一起，并在记录键和文件组之间进行映射。如上所述，所有更新都记录到特定文件组的增量日志文件中。这种设计比 Hive ACID 更高效，Hive ACID 必须将所有数据记录与所有基础文件合并才能处理查询。 Hudi 的设计预期基于键的快速更新插入和删除，因为它适用于文件组的增量日志，而不是整个数据集。 。 来源 对于理解至关重要，因为它是所有 Hudi 表元数据的真实事件日志来源。时间线存储在 文件夹中，或者在我们的例子中存储在存储桶中。事件会保留在时间线上，直到被删除。时间线适用于整个表以及文件组，从而可以通过将增量日志应用到原始基础文件来重建文件组。为了优化频繁写入/提交，Hudi 的设计使元数据相对于整个表的大小较小。 时间线 .hoodie 时间线上的新事件被保存到内部元数据表中，并作为一系列读时合并表实现，从而提供低写入放大。因此，Hudi 可以快速吸收元数据的快速变化。此外，元数据表使用 HFile 基本文件格式，通过一组索引键查找进一步优化性能，避免读取整个元数据表。作为表一部分的所有物理文件路径都包含在元数据中，以避免昂贵且耗时的云文件列表。 胡迪作家 Hudi 编写器促进了这样的架构：Hudi 充当高性能写入层，具有 ACID 事务支持，可实现非常快速的增量更改，例如更新和删除。 典型的 Hudi 架构依赖于 Spark 或 Flink 管道将数据传输到 Hudi 表。 Hudi 写入路径经过优化，比简单地将 Parquet 或 Avro 文件写入磁盘更有效。 Hudi 分析写入操作并将其分类为增量操作（ 、 、 ）或批量操作（ 、 、 、 ），然后应用必要的 。 insert upsert delete insert_overwrite insert_overwrite_table delete_partition bulk_insert 优化 Hudi 编写者还负责维护元数据。对于每条记录，都会写入提交时间和该记录唯一的序列号（这类似于 Kafka 偏移量），从而可以导出记录级别的更改。用户还可以在传入数据流中指定事件时间字段，并使用元数据和 Hudi 时间线跟踪它们。这可以显着改进流处理，因为 Hudi 包含每个记录的到达时间和事件时间，从而可以为复杂的流处理管道构建强大的 。 水印 胡迪读者 写入器和读取器之间的快照隔离允许从所有主要数据湖查询引擎（包括 Spark、Hive、Flink、Prest、Trino 和 Impala）一致地查询表快照。与 Parquet 和 Avro 一样，Hudi 表可以被 和 等读取为外部表。 Snowflake SQL Server Hudi 阅读器被开发为轻量级。只要有可能，就会使用特定于引擎的矢量化读取器和缓存，例如 Presto 和 Spark 中的那些。当 Hudi 必须合并查询的基本文件和日志文件时，Hudi 使用可溢出映射和延迟读取等机制来提高合并性能，同时还提供读取优化的查询。 Hudi 包含多种非常强大的增量查询功能。元数据是其核心，它允许将大型提交作为较小的块使用，并完全解耦数据的写入和增量查询。通过有效使用元数据，时间旅行只是另一个具有定义的起点和终点的增量查询。 Hudi 在任何给定时间点以原子方式将键映射到单个文件组，支持 Hudi 表上的完整 CDC 功能。正如上面 Hudi writers 部分所讨论的，每个表都由文件组组成，每个文件组都有自己包含的元数据。 胡迪万岁！ Hudi 的最大优势在于它摄取流数据和批量数据的速度。通过提供 功能，Hudi 执行任务的速度比重写整个表或分区快几个数量级。 upsert 为了利用 Hudi 的摄取速度，数据湖屋需要具有高 IOPS 和吞吐量的存储层。 MinIO 的可扩展性和高性能的结合正是 Hudi 所需要的。 MinIO 完全能够满足为实时企业数据湖提供支持所需的性能 — 在 GET 上实现了 325 GiB/s (349 GB/s)，在 PUT 上实现了 165 GiB/s (177 GB/s)。 32 个节点的现成 NVMe SSD。 最近的基准测试 活跃的企业 Hudi 数据湖存储大量小型 Parquet 和 Avro 文件。 MinIO 包括许多 ，可实现更快的数据湖。小对象与元数据内联保存，减少了读写 Hudi 元数据和索引等小文件所需的 IOPS。 小文件优化 模式是每个 Hudi 表的重要组成部分。 Hudi 可以强制执行模式，也可以允许 ，以便流数据管道可以在不中断的情况下进行调整。此外，Hudi 强制执行 schema-on-writer 以确保更改不会破坏管道。 Hudi 依赖 Avro 来存储、管理和发展表的架构。 模式演变 Hudi 为数据湖提供 。 Hudi 确保原子写入：提交以原子方式提交到时间线，并给出一个时间戳，该时间戳表示操作被视为发生的时间。 Hudi 将写入器、表和读取器进程之间的快照隔离开来，以便每个进程都对表的一致快照进行操作。 Hudi 通过写入器之间的乐观并发控制 (OCC) 以及表服务和写入器之间以及多个表服务之间的基于非阻塞 MVCC 的并发控制来解决此问题。 ACID 事务保证 Hudi 和 MinIO 教程 本教程将引导您完成 Spark、Hudi 和 MinIO 的设置，并介绍一些基本的 Hudi 功能。本教程基于 ，适用于云原生 MinIO 对象存储。 Apache Hudi Spark 指南 请注意，使用版本化存储桶会增加 Hudi 的一些维护开销。任何被删除的对象都会创建一个 。当 Hudi 使用 删除标记的数量会随着时间的推移而增加。正确配置 以清理这些删除标记非常重要，因为如果删除标记的数量达到 1000，列表操作可能会阻塞。Hudi 项目维护人员建议在使用生命周期规则一天后清理删除标记。 删除标记 Cleaner 实用程序清理文件时， 生命周期管理 先决条件 Apache Spark。 下载并安装 MinIO。记录控制台的 IP 地址、TCP 端口、访问密钥和秘密密钥。 下载并安装 MinIO 客户端。 下载并安装 下载 AWS 和 AWS Hadoop 库并将它们添加到您的类路径中，以便使用 S3A 来处理对象存储。 AWS： （或更高版本） aws-java-sdk:1.10.34 Hadoop： （或更高版本） hadoop-aws:2.7.3 Jar 文件，解压缩并将其复制到 。 下载 /opt/spark/jars 创建 MinIO 桶 使用 MinIO 客户端创建一个存储桶来存放 Hudi 数据： mc alias set myminio http:// mc mb myminio/hudi 使用 Hudi 启动 Spark 启动 Spark shell，并将 Hudi 配置为使用 MinIO 进行存储。确保使用 MinIO 设置配置 S3A 条目。 spark-shell \ --packages org.apache.hudi:hudi-spark3.3-bundle_2.12:0.12.0,org.apache.hadoop:hadoop-aws:3.3.4 \ --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \ --conf 'spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog' \ --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension' \ --conf 'spark.hadoop.fs.s3a.access.key= ' \ --conf 'spark.hadoop.fs.s3a.secret.key= '\ --conf 'spark.hadoop.fs.s3a.endpoint= :9000' \ --conf 'spark.hadoop.fs.s3a.path.style.access=true' \ --conf 'fs.s3a.signing-algorithm=S3SignerType' 然后，在 Spark 中初始化 Hudi。 import org.apache.hudi.QuickstartUtils._ import scala.collection.JavaConversions._ import org.apache.spark.sql.SaveMode._ import org.apache.hudi.DataSourceReadOptions._ import org.apache.hudi.DataSourceWriteOptions._ import org.apache.hudi.config.HoodieWriteConfig._ import org.apache.hudi.common.model.HoodieRecord 请注意，它将简化 Hudi 的重复使用来创建 。 外部配置文件 创建一个表 尝试一下并使用 Scala 创建一个简单的小型 Hudi 表。 Hudi DataGenerator 是一种基于 生成示例插入和更新的快速且简单的方法。 示例行程架构 val tableName = "hudi_trips_cow" val basePath = "s3a://hudi/hudi_trips_cow" val dataGen = new DataGenerator 向Hudi插入数据并将表写入MinIO 下面将生成新的行程数据，将它们加载到 DataFrame 中，并将我们刚刚创建的 DataFrame 作为 Hudi 表写入 MinIO。如果表已存在， 将覆盖并重新创建该表。行程数据依赖于记录键 ( )、分区字段 ( ) 和逻辑 ( ) 以确保每个分区的行程记录是唯一的。我们将使用默认的写入操作 。当您的工作负载没有更新时，您可以使用 或 ，这可能会更快。 mode(Overwrite) uuid region/country/city ts upsert insert bulk_insert val inserts = convertToStringList(dataGen.generateInserts(10)) val df = spark.read.json(spark.sparkContext.parallelize(inserts, 2)) df.write.format("hudi"). options(getQuickstartWriteConfigs). option(PRECOMBINE_FIELD_OPT_KEY, "ts"). option(RECORDKEY_FIELD_OPT_KEY, "uuid"). option(PARTITIONPATH_FIELD_OPT_KEY, "partitionpath"). option(TABLE_NAME, tableName). mode(Overwrite). save(basePath) 打开浏览器并使用您的访问密钥和秘密密钥通过 登录 MinIO。您将在存储桶中看到 Hudi 表。 http:// : 该存储桶还包含包含元数据的 路径以及包含数据的 和 路径。 .hoodie americas asia 查看元数据。这就是完成整个教程后我的 路径的样子。我们可以看到我在2022年9月13日星期二的9:02、10:37、10:48、10:52和10:56修改了表格。 .hoodie 查询数据 让我们将 Hudi 数据加载到 DataFrame 中并运行示例查询。 // spark-shell val tripsSnapshotDF = spark. read. format("hudi"). load(basePath) tripsSnapshotDF.createOrReplaceTempView("hudi_trips_snapshot") spark.sql("select fare, begin_lon, begin_lat, ts from hudi_trips_snapshot where fare > 20.0").show() spark.sql("select _hoodie_commit_time, _hoodie_record_key, _hoodie_partition_path, rider, driver, fare from hudi_trips_snapshot").show() 与胡迪一起穿越时空 不，我们不是在谈论 1988 年去看 音乐会。 Hootie and the Blowfish 每次写入 Hudi 表都会创建新的快照。将快照视为可用于时间旅行查询的表版本。 尝试一些时间旅行查询（您必须更改时间戳才能与您相关）。 spark.read. format("hudi"). option("as.of.instant", "2022-09-13 09:02:08.200"). load(basePath) 更新数据 这个过程与我们之前插入新数据时类似。为了展示 Hudi 更新数据的能力，我们将生成对现有行程记录的更新，将它们加载到 DataFrame 中，然后将 DataFrame 写入已保存在 MinIO 中的 Hudi 表中。 请注意，我们使用的是 保存模式。一般准则是使用 模式，除非您要创建新表，这样不会覆盖任何记录。使用 Hudi 的典型方法是实时摄取流数据，将它们附加到表中，然后编写一些逻辑，根据刚刚附加的内容合并和更新现有记录。或者，使用 模式写入会删除并重新创建该表（如果该表已存在）。 append append overwrite // spark-shell val updates = convertToStringList(dataGen.generateUpdates(10)) val df = spark.read.json(spark.sparkContext.parallelize(updates, 2)) df.write.format("hudi"). options(getQuickstartWriteConfigs). option(PRECOMBINE_FIELD_OPT_KEY, "ts"). option(RECORDKEY_FIELD_OPT_KEY, "uuid"). option(PARTITIONPATH_FIELD_OPT_KEY, "partitionpath"). option(TABLE_NAME, tableName). mode(Append). save(basePath) 查询数据会显示更新的行程记录。 增量查询 Hudi 可以使用增量查询提供自给定时间戳以来更改的记录流。我们需要做的就是提供一个开始时间，从该时间开始流式传输更改，以查看当前提交之前的更改，并且我们可以使用结束时间来限制流。 增量查询对于 Hudi 来说非常重要，因为它允许您在批量数据上构建流式管道。 // spark-shell // reload data spark. read. format("hudi"). load(basePath). createOrReplaceTempView("hudi_trips_snapshot") val commits = spark.sql("select distinct(_hoodie_commit_time) as commitTime from hudi_trips_snapshot order by commitTime").map(k => k.getString(0)).take(50) val beginTime = commits(commits.length - 2) // commit time we are interested in // incrementally query data val tripsIncrementalDF = spark.read.format("hudi"). option(QUERY_TYPE_OPT_KEY, QUERY_TYPE_INCREMENTAL_OPT_VAL). option(BEGIN_INSTANTTIME_OPT_KEY, beginTime). load(basePath) tripsIncrementalDF.createOrReplaceTempView("hudi_trips_incremental") spark.sql("select `_hoodie_commit_time`, fare, begin_lon, begin_lat, ts from hudi_trips_incremental where fare > 20.0").show() 时间点查询 Hudi可以查询截至特定时间和日期的数据。 // spark-shell val beginTime = "000" // Represents all commits > this time. val endTime = commits(commits.length - 2) // commit time we are interested in //incrementally query data val tripsPointInTimeDF = spark.read.format("hudi"). option(QUERY_TYPE_OPT_KEY, QUERY_TYPE_INCREMENTAL_OPT_VAL). option(BEGIN_INSTANTTIME_OPT_KEY, beginTime). option(END_INSTANTTIME_OPT_KEY, endTime). load(basePath) tripsPointInTimeDF.createOrReplaceTempView("hudi_trips_point_in_time") spark.sql("select `_hoodie_commit_time`, fare, begin_lon, begin_lat, ts from hudi_trips_point_in_time where fare > 20.0").show() 使用软删除删除数据 Hudi 支持两种不同的删除记录的方式。软删除保留记录键并清空所有其他字段的值。软删除会持久保存在 MinIO 中，并且只能使用硬删除从数据湖中删除。 // spark-shell spark. read. format("hudi"). load(basePath). createOrReplaceTempView("hudi_trips_snapshot") // fetch total records count spark.sql("select uuid, partitionpath from hudi_trips_snapshot").count() spark.sql("select uuid, partitionpath from hudi_trips_snapshot where rider is not null").count() // fetch two records for soft deletes val softDeleteDs = spark.sql("select * from hudi_trips_snapshot").limit(2) // prepare the soft deletes by ensuring the appropriate fields are nullified val nullifyColumns = softDeleteDs.schema.fields. map(field => (field.name, field.dataType.typeName)). filter(pair => (!HoodieRecord.HOODIE_META_COLUMNS.contains(pair._1) && !Array("ts", "uuid", "partitionpath").contains(pair._1))) val softDeleteDf = nullifyColumns. foldLeft(softDeleteDs.drop(HoodieRecord.HOODIE_META_COLUMNS: _*))( (ds, col) => ds.withColumn(col._1, lit(null).cast(col._2))) // simply upsert the table after setting these fields to null softDeleteDf.write.format("hudi"). options(getQuickstartWriteConfigs). option(OPERATION_OPT_KEY, "upsert"). option(PRECOMBINE_FIELD_OPT_KEY, "ts"). option(RECORDKEY_FIELD_OPT_KEY, "uuid"). option(PARTITIONPATH_FIELD_OPT_KEY, "partitionpath"). option(TABLE_NAME, tableName). mode(Append). save(basePath) // reload data spark. read. format("hudi"). load(basePath). createOrReplaceTempView("hudi_trips_snapshot") // This should return the same total count as before spark.sql("select uuid, partitionpath from hudi_trips_snapshot").count() // This should return (total - 2) count as two records are updated with nulls spark.sql("select uuid, partitionpath from hudi_trips_snapshot where rider is not null").count() 使用硬删除删除数据 相比之下，硬删除就是我们所说的删除。记录键和关联字段将从表中删除。 // spark-shell // fetch total records count spark.sql("select uuid, partitionpath from hudi_trips_snapshot").count() // fetch two records to be deleted val ds = spark.sql("select uuid, partitionpath from hudi_trips_snapshot").limit(2) // issue deletes val deletes = dataGen.generateDeletes(ds.collectAsList()) val hardDeleteDf = spark.read.json(spark.sparkContext.parallelize(deletes, 2)) hardDeleteDf.write.format("hudi"). options(getQuickstartWriteConfigs). option(OPERATION_OPT_KEY,"delete"). option(PRECOMBINE_FIELD_OPT_KEY, "ts"). option(RECORDKEY_FIELD_OPT_KEY, "uuid"). option(PARTITIONPATH_FIELD_OPT_KEY, "partitionpath"). option(TABLE_NAME, tableName). mode(Append). save(basePath) // run the same read query as above. val roAfterDeleteViewDF = spark. read. format("hudi"). load(basePath) roAfterDeleteViewDF.registerTempTable("hudi_trips_snapshot") // fetch should return (total - 2) records spark.sql("select uuid, partitionpath from hudi_trips_snapshot").count() 插入覆盖 当数据湖获得更新现有数据的能力时，它就成为数据湖屋。我们将生成一些新的行程数据，然后覆盖现有数据。此操作比 更快，其中 Hudi 会立即为您计算整个目标分区。在这里，我们指定配置，以便绕过 为您执行的自动索引、预组合和重新分区。 upsert upsert // spark-shell spark. read.format("hudi"). load(basePath). select("uuid","partitionpath"). sort("partitionpath","uuid"). show(100, false) val inserts = convertToStringList(dataGen.generateInserts(10)) val df = spark. read.json(spark.sparkContext.parallelize(inserts, 2)). filter("partitionpath = 'americas/united_states/san_francisco'") df.write.format("hudi"). options(getQuickstartWriteConfigs). option(OPERATION.key(),"insert_overwrite"). option(PRECOMBINE_FIELD.key(), "ts"). option(RECORDKEY_FIELD.key(), "uuid"). option(PARTITIONPATH_FIELD.key(), "partitionpath"). option(TBL_NAME.key(), tableName). mode(Append). save(basePath) // Should have different keys now for San Francisco alone, from query before. spark. read.format("hudi"). load(basePath). select("uuid","partitionpath"). sort("partitionpath","uuid"). show(100, false) 演进表架构和分区 架构演变允许您更改 Hudi 表的架构以适应数据随时间发生的变化。 以下是如何查询和发展架构和分区的一些示例。更深入的讨论，请参阅 。请注意，如果您运行这些命令，它们将更改您的 Hudi 表架构以与本教程不同。 Schema Evolution |阿帕奇胡迪 -- Alter table name ALTER TABLE oldTableName RENAME TO newTableName -- Alter table add columns ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*) -- Alter table column type ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType -- Alter table properties ALTER TABLE tableIdentifier SET TBLPROPERTIES (key = 'value') #Alter table examples --rename to: ALTER TABLE hudi_cow_nonpcf_tbl RENAME TO hudi_cow_nonpcf_tbl2; --add column: ALTER TABLE hudi_cow_nonpcf_tbl2 add columns(remark string); --change column: ALTER TABLE hudi_cow_nonpcf_tbl2 change column uuid uuid bigint; --set properties; alter table hudi_cow_nonpcf_tbl2 set tblproperties (hoodie.keep.max.commits = '10'); 目前， 仅适用于文件系统，因为它基于文件系统表路径。 SHOW partitions 本教程使用 Spark 来展示 Hudi 的功能。但是，Hudi 可以支持多种表类型/查询类型，并且可以从 Hive、Spark、Presto 等查询引擎查询 Hudi 表。 Hudi 项目有一个 ，在基于 Docker 的设置上展示了所有这些内容，所有相关系统都在本地运行。 演示视频 叱！叱！让我们在 MinIO 上构建 Hudi 数据湖吧！ Apache Hudi 是第一个数据湖开放表格式，在流式架构中值得考虑。 充满活力，越来越重视用 Hudi/对象存储替代 Hadoop/HDFS，以实现云原生流数据湖。使用 MinIO 进行 Hudi 存储为多云数据湖和分析铺平了道路。 MinIO 包括 ，可在本地、公共/私有云和边缘等位置之间同步数据，从而实现企业所需的出色功能，例如地理负载平衡和快速热热故障转移。 Hudi 社区和生态系统 主动-主动复制 今天在 MinIO 上试试 Hudi。如果您有任何疑问或想要分享技巧，请通过 联系。 我们的 Slack 频道 也发布 。 在这里