官宣！AWS Athena正式可查詢Apache Hudi數據集

1. 引入

Apache Hudi是一個開源的增量數據處理框架，提供了行級insert、update、upsert、delete的細粒度處理能力（Upsert表示如果數據集中存在記錄就更新；否則插入）。

Hudi處理數據插入和更新，不會創建太多的小文件(小文件會導致查詢端性能降低)，Apache Hudi自動管理及合並小文件，讓其保持指定大小，這避免了自建解決方案來監控和重寫小文件為大文件。

Hudi數據集在如下場景下非常適用

• 使用GDPR和CCPA法規來刪除用戶個人信息或修改個人信息用途。
• 處理傳感器或IoT設備的流式數據，涉及數據插入和更新。
• 實現CDC系統

Hudi使用開放的數據格式管理S3的數據集。現在Athena可以查詢Hudi數據集，但暫還不支持寫入，Athena使用Apache Hudi 0.5.2-incubating版本，0.5.2-incubating版本信息可參考這里

2. Hudi數據集類型

Hudi數據集有如下類型

• Copy on Write (CoW) – 使用Parquet列式存儲，每次更新將會創建一個新版本。

• Merge on Read (MoR) – 使用Parquet列式 + Avro行式存儲，更新將會寫入delta日志文件，后面將會和Parquet列式文件進行壓縮生成新版本列式文件。

對於CoW數據集，對記錄更新時，包含記錄的文件將會被重寫；對於MoR數據集，對記錄更新時，Hudi僅僅只會寫更新的值。因此MoR更適合重寫的場景，CoW更適合重讀場景（數據很少變更）。

Hudi提供了三種邏輯視圖來訪問數據：

• Read-optimized 視圖 – 提供CoW表最新提交的數據集和MoR表最新壓縮的數據集，均讀取Parquet文件。
• Incremental 視圖 – 提供CoW表中兩次提交的變更流，便於下游ETL作業。
• Real-time 視圖 – 提供MoR表最新提交的數據，在查詢時合並列式和行式文件。

現在Athena只支持Read-optimized視圖，這提供了更好的查詢性能但未包含最新的delta提交。關於數據集類型做的tradeoff，可以參考Hudi文檔Storage Types & Views 。

3. 考慮及限制

• Athena對Hudi數據集僅支持查詢Read-optimized視圖
  • 對於CoW類型，Athena支持快照查詢；
  • 對於MoR類型，Athena支持讀優化查詢；
• Athena對Hudi數據集不支持CTAS 或 INSERT INTO，更多關於如何寫入Hudi數據集，可參考
  • Amazon EMR 發布指南中玩轉Hudi數據集
  • Apache Hudi文檔：寫Hudi表
• Athena對Hudi表不支持使用MSCK REPAIR TABLE。如果需要加載非Glue創建的Hudi表，請使用ALTER TABLE ADD PARTITION

4. 創建Hudi表

本部分將提供Athena中創建分區和非分區Hudi表的建表示例。

如果已經在AWS Glue中創建了Hudi表，那么可以直接使用Athena查詢。如果在Athena中創建Hudi表，在查詢之前必須運行ALTER TABLE ADD PARTITION 來加載數據。

4.1 Copy on Write (CoW)建表示例

4.1.1 非分區CoW表

下面示例會在Athena中創建非分區CoW表

CREATE EXTERNAL TABLE `non_partition_cow`(
  `_hoodie_commit_time` string,
  `_hoodie_commit_seqno` string,
  `_hoodie_record_key` string,
  `_hoodie_partition_path` string,
  `_hoodie_file_name` string,
  `event_id` string,
  `event_time` string,
  `event_name` string,
  `event_guests` int,
  `event_type` string)
ROW FORMAT SERDE
  'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
STORED AS INPUTFORMAT
  'org.apache.hudi.hadoop.HoodieParquetInputFormat'
OUTPUTFORMAT
  'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION
  's3://bucket/folder/non_partition_cow'

4.1.2 分區CoW表

下面示例會在Athena中創建分區CoW表

CREATE EXTERNAL TABLE `partition_cow`(
  `_hoodie_commit_time` string, 
  `_hoodie_commit_seqno` string, 
  `_hoodie_record_key` string, 
  `_hoodie_partition_path` string, 
  `_hoodie_file_name` string, 
  `event_id` string, 
  `event_time` string, 
  `event_name` string, 
  `event_guests` int)
PARTITIONED BY ( 
  `event_type` string)
ROW FORMAT SERDE 
  'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe' 
STORED AS INPUTFORMAT 
  'org.apache.hudi.hadoop.HoodieParquetInputFormat' 
OUTPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION
  's3://bucket/folder/partition_cow'

下面ALTER TABLE ADD PARTITION示例會添加兩個分區到partition_cow 表

ALTER TABLE partition_cow ADD
  PARTITION (event_type = 'one') LOCATION 's3://bucket/folder/partition_cow/one/'
  PARTITION (event_type = 'two') LOCATION 's3://bucket/folder/partition_cow/two/'

4.2 Merge on Read (MoR)建表示例

Hudi對於MoR類型將會在Hive Metastore中創建兩張表：一張由你指定的表，可提供Read-optimized視圖，另一張以_rt結尾的表，可提供Real-time視圖。然而當你在Athena創建MoR表時，也只能查詢read-optimized視圖（real-time視圖支持社區正在進行代碼Review，不久后可用）。

4.2.1 非分區MoR表

下面示例會在Athena中創建非分區MoR表

CREATE EXTERNAL TABLE `nonpartition_mor_ro`(
  `_hoodie_commit_time` string, 
  `_hoodie_commit_seqno` string, 
  `_hoodie_record_key` string, 
  `_hoodie_partition_path` string, 
  `_hoodie_file_name` string, 
  `event_id` string, 
  `event_time` string, 
  `event_name` string, 
  `event_guests` int, 
  `event_type` string)
ROW FORMAT SERDE 
  'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe' 
STORED AS INPUTFORMAT 
  'org.apache.hudi.hadoop.HoodieParquetInputFormat' 
OUTPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION
  's3://bucket/folder/nonpartition_mor'

4.2.2 分區MoR表

下面示例會在Athena中創建分區MoR表

CREATE EXTERNAL TABLE `partition_mor_ro`(
  `_hoodie_commit_time` string, 
  `_hoodie_commit_seqno` string, 
  `_hoodie_record_key` string, 
  `_hoodie_partition_path` string, 
  `_hoodie_file_name` string, 
  `event_id` string, 
  `event_time` string, 
  `event_name` string, 
  `event_guests` int)
PARTITIONED BY ( 
  `event_type` string)
ROW FORMAT SERDE 
  'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe' 
STORED AS INPUTFORMAT 
  'org.apache.hudi.hadoop.HoodieParquetInputFormat' 
OUTPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION
  's3://bucket/folder/partition_mor'

下面ALTER TABLE ADD PARTITION示例會添加兩個分區到partition_mor_ro表

ALTER TABLE partition_cow ADD
  PARTITION (event_type = 'one') LOCATION 's3://bucket/folder/partition_mor/one/'
  PARTITION (event_type = 'two') LOCATION 's3://bucket/folder/partition_mor/two/'