vlambda博客
学习文章列表

速度！Apache Hudi又双叕被国内顶级云服务提供商集成了！

Apache Hudi 在 HDFS 的数据集上提供了插入更新和增量拉取的流原语。

一般来说，我们会将大量数据存储到 HDFS，新数据增量写入，而旧数据鲜有改动，特别是在经过数据清洗，放入数据仓库的场景。而且在数据仓库如 hive 中，对于 update 的支持非常有限，计算昂贵。另一方面，若是有仅对某段时间内新增数据进行分析的场景，则 hive、presto、hbase 等也未提供原生方式，而是需要根据时间戳进行过滤分析。

在此需求下，Hudi 可以提供这两种需求的实现。第一个是对 record 级别的更新，另一个是仅对增量数据的查询。且 Hudi 提供了对 Hive、presto、Spark 的支持，可以直接使用这些组件对 Hudi 管理的数据进行查询。

Hudi 是一个通用的大数据存储系统，主要特性：

  • 摄取和查询引擎之间的快照隔离，包括 Apache Hive、Presto 和 Apache Spark。

  • 支持回滚和存储点，可以恢复数据集。

  • 自动管理文件大小和布局，以优化查询性能准实时摄取，为查询提供最新数据。

  • 实时数据和列数据的异步压缩。

时间轴

在它的核心，Hudi 维护一条包含在不同的即时时间所有对数据集操作的时间轴，从而提供了从不同时间点出发得到不同的视图下的数据集。

Hudi 即时包含以下组件：

  • 操作类型：对数据集执行的操作类型。

  • 即时时间：即时时间通常是一个时间戳（例如：20190117010349），该时间戳按操作开始时间的顺序单调增加。

  • 状态：即时的状态。

文件组织

Hudi 将 DFS 上的数据集组织到 基本路径下的目录结构中。数据集分为多个分区，这些分区是包含该分区的数据文件的文件夹，这与 Hive 表非常相似。

每个分区被相对于基本路径的特定 分区路径区分开来。

在每个分区内，文件被组织为 文件组，由 文件id唯一标识。

每个文件组包含多个 文件切片，其中每个切片包含在某个提交/压缩即时时间生成的基本列文件 *.parquet以及一组日志文件 *.log*，该文件包含自生成基本文件以来对基本文件的插入/更新。

Hudi 采用 MVCC 设计，其中压缩操作将日志和基本文件合并以产生新的文件片，而清理操作则将未使用的/较旧的文件片删除以回收 DFS 上的空间。

Hudi 通过索引机制将给定的 hoodie 键（记录键+分区路径）映射到文件组，从而提供了高效的 Upsert。

一旦将记录的第一个版本写入文件，记录键和 文件组/ 文件id之间的映射就永远不会改变。简而言之，映射的文件组包含一组记录的所有版本。

存储类型

Hudi 支持以下存储类型：

  • 写时复制：仅使用列文件格式（例如 parquet）存储数据。通过在写入过程中执行同步合并以更新版本并重写文件。

  • 读时合并：使用列式（例如 parquet）+ 基于行（例如 avro）的文件格式组合来存储数据。更新记录到增量文件中，然后进行同步或异步压缩以生成列文件的新版本。

下表总结了这两种存储类型之间的权衡：

权衡 写时复制 读时合并
数据延迟 更高 更低
更新代价(I/O) 更高（重写整个parquet文件） 更低（追加到增量日志）
Parquet文件大小 更小（高更新代价（I/o）） 更大（低更新代价）
写放大 更高 更低（取决于压缩策略）

Hudi 对 EMR 底层存储支持

  • HDFS

  • COS

安装 Hudi

进入 EMR 购买页，选择【产品版本】为 EMR-V2.2.0，选择【可选组件】为【hudi 0.5.1】。hudi 组件默认安装在 master 和 router 节点上。

! hudi 组件依赖 hive 和 spark 组件， 如果选择安装 hudi 组件，EMR 将自动安装 hive 和 spark 组件。

使用示例

可参考 hudi 官网示例：

  1. 登录 master 节点，切换为 hadoop 用户。

  2. 加载 spark 配置。

 
   
   
 
    
  
    
    
  
  1. cd /usr/local/service/hudi
    2. 
  
    
    
  
  2. ln -s /usr/local/service/spark/conf/spark-defaults.conf /usr/local/service/hudi/demo/config/spark-defaults.conf
    3.

上传配置到 hdfs：

 
   
   
 
    
  
    
    
  
  1. hdfs dfs -mkdir -p /hudi/config
    2. 
  
    
    
  
  2. hdfs dfs -copyFromLocal demo/config/* /hudi/config/
    3.

  1. 修改 kafka 数据源。

 
   
   
 
    
  
    
    
  
  1. /usr/local/service/apps/hudi-0.5.1/demo/config/kafka-source.properties
    2. 
  
    
    
  
  2. bootstrap.servers=kafka_ip:kafka_port
    3.

上传第一批次数据：

 
   
   
 
    
  
    
    
  
  1. cat demo/data/batch_1.json | kafkacat -b [kafka_ip] -t stock_ticks -P
    2.

  1. 摄取第一批数据。

 
   
   
 
    
  
    
    
  
  1. spark-submit --class org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer --master yarn ./hudi-utilities-bundle_2.11-0.5.1-incubating.jar --table-type COPY_ON_WRITE --source-class org.apache.hudi.utilities.sources.JsonKafkaSource --source-ordering-field ts --target-base-path /usr/hive/warehouse/stock_ticks_cow --target-table stock_ticks_cow --props /hudi/config/kafka-source.properties --schemaprovider-class org.apache.hudi.utilities.schema.FilebasedSchemaProvider
    2. 
  
    
    
  
  2. spark-submit --class org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer --master yarn ./hudi-utilities-bundle_2.11-0.5.1-incubating.jar --table-type MERGE_ON_READ --source-class org.apache.hudi.utilities.sources.JsonKafkaSource --source-ordering-field ts --target-base-path /usr/hive/warehouse/stock_ticks_mor --target-table stock_ticks_mor --props /hudi/config/kafka-source.properties --schemaprovider-class org.apache.hudi.utilities.schema.FilebasedSchemaProvider --disable-compaction
    3.

  1. 查看 hdfs 数据。

 
   
   
 
    
  
    
    
  
  1.  hdfs dfs -ls /usr/hive/warehouse/
    2.

  1. 同步 hive 元数据。

 
   
   
 
    
  
    
    
  
  1. bin/run_sync_tool.sh --jdbc-url jdbc:hive2://[hiveserver2_ip:hiveserver2_port] --user hadoop --pass [password] --partitioned-by dt --base-path /usr/hive/warehouse/stock_ticks_cow --database default --table stock_ticks_cow
    2. 
  
    
    
  
  2. bin/run_sync_tool.sh --jdbc-url jdbc:hive2://[hiveserver2_ip:hiveserver2_port] --user hadoop --pass [password]--partitioned-by dt --base-path /usr/hive/warehouse/stock_ticks_mor --database default --table stock_ticks_mor --skip-ro-suffix
    3.

  1. 使用计算引擎查询数据。

  • hive 引擎

 
   
   
 
    
  
    
    
  
  1. beeline -u jdbc:hive2://[hiveserver2_ip:hiveserver2_port] -n hadoop --hiveconf hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat --hiveconf hive.stats.autogather=false
    2.

或者 spark 引擎

 
   
   
 
    
  
    
    
  
  1. spark-sql --master yarn --conf spark.sql.hive.convertMetastoreParquet=false
    2.

hive/spark 引擎执行如下 sql 语句：

 
   
   
 
    
  
    
    
  
  1. select symbol, max(ts) from stock_ticks_cow group by symbol HAVING symbol = 'GOOG';
    2. 
  
    
    
  
  2. select `_hoodie_commit_time`, symbol, ts, volume, open, close from stock_ticks_cow where symbol = 'GOOG';
    3. 
  
    
    
  
  3. select symbol, max(ts) from stock_ticks_mor group by symbol HAVING symbol = 'GOOG';
    4. 
  
    
    
  
  4. select `_hoodie_commit_time`, symbol, ts, volume, open, close from stock_ticks_mor where symbol = 'GOOG';
    5. 
  
    
    
  
  5. select symbol, max(ts) from stock_ticks_mor_rt group by symbol HAVING symbol = 'GOOG';
    6. 
  
    
    
  
  6. select `_hoodie_commit_time`, symbol, ts, volume, open, close from stock_ticks_mor_rt where symbol = 'GOOG';
    7.

  • 进入 presto 引擎

 
   
   
 
    
  
    
    
  
  1. /usr/local/service/presto-client/presto --server localhost:9000 --catalog hive --schema default --user Hadoop
    2.

presto 查询有下划线的字段需要用双引号，例如 "_hoodie_commit_time"，执行如下 sql 语句：

 
   
   
 
    
  
    
    
  
  1. select symbol, max(ts) from stock_ticks_cow group by symbol HAVING symbol = 'GOOG';
    2. 
  
    
    
  
  2. select "_hoodie_commit_time", symbol, ts, volume, open, close from stock_ticks_cow where symbol = 'GOOG';
    3. 
  
    
    
  
  3. select symbol, max(ts) from stock_ticks_mor group by symbol HAVING symbol = 'GOOG';
    4. 
  
    
    
  
  4. select "_hoodie_commit_time", symbol, ts, volume, open, close from stock_ticks_mor where symbol = 'GOOG';
    5. 
  
    
    
  
  5. select symbol, max(ts) from stock_ticks_mor_rt group by symbol HAVING symbol = 'GOOG';
    6.

  1. 上传第二批数据。

 
   
   
 
    
  
    
    
  
  1. cat demo/data/batch_2.json | kafkacat -b 10.0.1.70 -t stock_ticks -P
    2.

  1. 摄取第二批增量数据。

 
   
   
 
    
  
    
    
  
  1. spark-submit --class org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer --master yarn ./hudi-utilities-bundle_2.11-0.5.1-incubating.jar --table-type COPY_ON_WRITE --source-class org.apache.hudi.utilities.sources.JsonKafkaSource --source-ordering-field ts --target-base-path /usr/hive/warehouse/stock_ticks_cow --target-table stock_ticks_cow --props /hudi/config/kafka-source.properties --schemaprovider-class org.apache.hudi.utilities.schema.FilebasedSchemaProvider
    2. 
  
    
    
  
  2. spark-submit --class org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer --master yarn ./hudi-utilities-bundle_2.11-0.5.1-incubating.jar --table-type MERGE_ON_READ --source-class org.apache.hudi.utilities.sources.JsonKafkaSource --source-ordering-field ts --target-base-path /usr/hive/warehouse/stock_ticks_mor --target-table stock_ticks_mor --props /hudi/config/kafka-source.properties --schemaprovider-class org.apache.hudi.utilities.schema.FilebasedSchemaProvider --disable-compaction
    3.

  1. 查询增量数据，查询方法同步骤7。

  2. 使用 hive-cli 工具。

 
   
   
 
    
  
    
    
  
  1.  cli/bin/hudi-cli.sh
    2. 
  
    
    
  
  2. connect --path /usr/hive/warehouse/stock_ticks_mor
    3. 
  
    
    
  
  3. compactions show all
    4. 
  
    
    
  
  4. compaction schedule
    5. 
  
    
    
  
  5. 合并执行计划
    6. 
  
    
    
  
  6.  compaction run --compactionInstant [requestID] --parallelism 2 --sparkMemory 1G --schemaFilePath /hudi/config/schema.avsc --retry 1
    7.

  1. 使用 tez/spark 引擎查询。

 
   
   
 
    
  
    
    
  
  1. beeline -u jdbc:hive2://[hiveserver2_ip:hiveserver2_port] -n hadoop --hiveconf hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat --hiveconf hive.stats.autogather=false
    2. 
  
    
    
  
  2. set hive.execution.engine=tez;
    3. 
  
    
    
  
  3. set hive.execution.engine=spark;
    4.

然后执行 sql 查询，可参考步骤7。

与对象存储结合使用

与 hdfs 类似，需要在存储路径前加上 cosn://[bucket]。参考如下操作：

 
   
   
 
    
  
    
    
  
  1. bin/kafka-server-start.sh config/server.properties &
    2. 
  
    
    
  
  2. cat demo/data/batch_1.json | kafkacat -b kafkaip -t stock_ticks -P
    3. 
  
    
    
  
  3. cat demo/data/batch_2.json | kafkacat -b kafkaip -t stock_ticks -P
    4. 
  
    
    
  
  4. kafkacat -b kafkaip -L
    5. 
  
    
    
  
  5. hdfs dfs -mkdir -p cosn://[bucket]/hudi/config
    6. 
  
    
    
  
  6. hdfs dfs -copyFromLocal demo/config/* cosn://[bucket]/hudi/config/
    7. 
  
    
    
  

    8. 
  
    
    
  

    9. 
  
    
    
  
  9. spark-submit --class org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer --master yarn ./hudi-utilities-bundle_2.11-0.5.1-incubating.jar --table-type COPY_ON_WRITE --source-class org.apache.hudi.utilities.sources.JsonKafkaSource --source-ordering-field ts --target-base-path cosn://[bucket]/usr/hive/warehouse/stock_ticks_cow --target-table stock_ticks_cow --props cosn://[bucket]/hudi/config/kafka-source.properties --schemaprovider-class org.apache.hudi.utilities.schema.FilebasedSchemaProvider
    10. 
  
    
    
  

    11. 
  
    
    
  

    12. 
  
    
    
  
  12. spark-submit --class org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer --master yarn ./hudi-utilities-bundle_2.11-0.5.1-incubating.jar --table-type MERGE_ON_READ --source-class org.apache.hudi.utilities.sources.JsonKafkaSource --source-ordering-field ts --target-base-path cosn://[bucket]/usr/hive/warehouse/stock_ticks_mor --target-table stock_ticks_mor --props cosn://[bucket]/hudi/config/kafka-source.properties --schemaprovider-class org.apache.hudi.utilities.schema.FilebasedSchemaProvider --disable-compaction
    13. 
  
    
    
  

    14. 
  
    
    
  

    15. 
  
    
    
  
  15. bin/run_sync_tool.sh --jdbc-url jdbc:hive2://[hiveserver2_ip:hiveserver2_port] --user hadoop --pass isd@cloud --partitioned-by dt --base-path cosn://[bucket]/usr/hive/warehouse/stock_ticks_cow --database default --table stock_ticks_cow
    16. 
  
    
    
  

    17. 
  
    
    
  
  17. bin/run_sync_tool.sh --jdbc-url jdbc:hive2://[hiveserver2_ip:hiveserver2_port] --user hadoop --pass hive --partitioned-by dt --base-path cosn://[bucket]/usr/hive/warehouse/stock_ticks_mor --database default --table stock_ticks_mor --skip-ro-suffix
    18. 
  
    
    
  

    19. 
  
    
    
  

    20. 
  
    
    
  
  20. beeline -u jdbc:hive2://[hiveserver2_ip:hiveserver2_port] -n hadoop --hiveconf hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat --hiveconf hive.stats.autogather=false
    21. 
  
    
    
  

    22. 
  
    
    
  
  22. spark-sql --master yarn --conf spark.sql.hive.convertMetastoreParquet=false
    23. 
  
    
    
  

    24. 
  
    
    
  

    25. 
  
    
    
  
  25. hivesqls:
    26. 
  
    
    
  
  26. select symbol, max(ts) from stock_ticks_cow group by symbol HAVING symbol = 'GOOG';
    27. 
  
    
    
  
  27. select `_hoodie_commit_time`, symbol, ts, volume, open, close from stock_ticks_cow where symbol = 'GOOG';
    28. 
  
    
    
  
  28. select symbol, max(ts) from stock_ticks_mor group by symbol HAVING symbol = 'GOOG';
    29. 
  
    
    
  
  29. select `_hoodie_commit_time`, symbol, ts, volume, open, close from stock_ticks_mor where symbol = 'GOOG';
    30. 
  
    
    
  
  30. select symbol, max(ts) from stock_ticks_mor_rt group by symbol HAVING symbol = 'GOOG';
    31. 
  
    
    
  
  31. select `_hoodie_commit_time`, symbol, ts, volume, open, close from stock_ticks_mor_rt where symbol = 'GOOG';
    32. 
  
    
    
  

    33. 
  
    
    
  
  33. prestosqls:
    34. 
  
    
    
  
  34. /usr/local/service/presto-client/presto --server localhost:9000 --catalog hive --schema default --user Hadoop
    35. 
  
    
    
  
  35. select symbol, max(ts) from stock_ticks_cow group by symbol HAVING symbol = 'GOOG';
    36. 
  
    
    
  
  36. select "_hoodie_commit_time", symbol, ts, volume, open, close from stock_ticks_cow where symbol = 'GOOG';
    37. 
  
    
    
  
  37. select symbol, max(ts) from stock_ticks_mor group by symbol HAVING symbol = 'GOOG';
    38. 
  
    
    
  
  38. select "_hoodie_commit_time", symbol, ts, volume, open, close from stock_ticks_mor where symbol = 'GOOG';
    39. 
  
    
    
  
  39. select symbol, max(ts) from stock_ticks_mor_rt group by symbol HAVING symbol = 'GOOG';
    40. 
  
    
    
  
  40. select "_hoodie_commit_time", symbol, ts, volume, open, close from stock_ticks_mor_rt where symbol = 'GOOG';
    41. 
  
    
    
  

    42. 
  
    
    
  

    43. 
  
    
    
  
  43. cli/bin/hudi-cli.sh
    44. 
  
    
    
  
  44. connect --path cosn://[bucket]/usr/hive/warehouse/stock_ticks_mor
    45. 
  
    
    
  
  45. compactions show all
    46. 
  
    
    
  
  46. compaction schedule
    47. 
  
    
    
  
  47. compaction run --compactionInstant [requestid] --parallelism 2 --sparkMemory 1G --schemaFilePath cosn://[bucket]/hudi/config/schema.avsc --retry 1
    48.


推荐阅读

[5.05] 暗喻幻想港中35周年限定版1.03+DLC+金手指+王中王存档 [5.05] 暗喻幻想港中35周年限定版1.03+DLC+金手指+王中王存档
数据仓库的ETL、OLAP和BI应用 数据仓库的ETL、OLAP和BI应用

相关文章