长亭百川云 - 文章详情

爱奇艺数据湖实战 - Hive数仓平滑入湖

爱奇艺技术产品团队

58

2024-07-13

爱奇艺基于 Hive 构建了传统的离线数据仓库,支持了公司运营决策、用户增长、视频推荐、会员、广告等业务需求。近几年,随着业务对数据实时性的更高要求。我们引入了基于 Iceberg 的数据湖技术,大幅提升数据查询性能及整体流通效率。从性能和成本角度考虑,将现有的Hive表迁移到数据湖是必要的。然而多年来,大数据平台上已经积累了数百 PB 的 Hive 数据,如何将 Hive 迁移到数据湖,成为我们面临的一大挑战。本文介绍了爱奇艺从 Hive 平滑迁移到 Iceberg 数据湖的技术方案,帮助业务加速数据流程,提效增收。

01

   Hive VS Iceberg

Hive 是一个基于 Hadoop 的数据仓库和分析平台,提供了类似 SQL 的语言,支持复杂的数据处理和分析。

Iceberg 是一个开源的数据表格式,旨在提供可扩展、稳定和高效的表格存储,以支持分析性工作负载。Iceberg 提供了类似传统数据库的事务性保证和数据一致性,并支持复杂的数据操作,如更新、删除等。

表 1-1 分别列出了 Hive 和 Iceberg 在时效性、查询性能等方面的比较情况:

表 1-1 Hive 和 Iceberg 的对比

切换到 Iceberg 可以提高数据处理的效率和可靠性,为复杂的数据操作提供更好的支持,目前已接入广告、会员、Venus 日志、审核等十几个业务。关于爱奇艺 Iceberg 实践的更多细节,可阅读之前的系列文章(见文末引用)。

02

   Hive 存量数据平滑切换 Iceberg

Iceberg 相比 Hive 有诸多优势,可是业务的数据已经运行在 Hive 环境中,业务不希望投入大量的人力修改存量的任务。我们调研了业界常用切换方法 [1],在数据湖平台上提供了自助式 Hive 平滑切换 Iceberg 的能力,本节将阐述具体实现方案。

1. 查询****兼容性

在实际切换前,我们验证了 Spark 对 Hive 和 Iceberg 的兼容性。

Spark 对 Hive 和 Iceberg 表的查询、写入语法基本是相同的,对 Hive 表的查询 SQL 语句无需修改即可查询 Iceberg 表。

但 Iceberg 与 Hive 在 DDL 方面存在较大差异,主要表现在对表结构进行修改的处理方式上,详细信息如表 2-1 所述。实际 schema 与数据文件的 schema 需要一一对应,否则会影响数据的查询,因此对于 DDL 语句的处理应该更为谨慎,不建议将这类 DDL 语句与任务绑定在一起。

表 2-1 Hive 和 Iceberg 语法兼容性对比

2. 业界切换方案

2**.1 业务双写切换**

业务复制现有的pipeline,实现Hive、Iceberg双写。待新旧通路对数一致后,切换到 Iceberg 通路,并下线原有的通路。该方案需要业务投入人力进行开发、对数,耗时耗力。

2**.2 原地切换,客户端停写**

如果业务允许停写一段时间进行切换,则可以用如下一些方式:

  • Spark migrate procedure 是 Iceberg 官方提供的函数,可将一个 Hive 表原地切换为 Iceberg,示例如下:

CALL catalog_name.system.migrate('db.sample');

该程序不会修改原始数据,仅会扫描原表的数据然后构建 Iceberg 元信息,引用原始的文件。因而 migrate 程序执行速度非常快,但存量数据无法利用文件索引等特性加速查询。如希望存量数据也加速,可以使用 Spark 的rewrite_data_files 方法重写历史数据。

migrate 程序并不会将 Hive 表删除,而是将其重命名为 sample__BACKUP__,此处 __BACKUP__ 后缀是硬编码的,如果需要回滚可将新建的 Iceberg 表 Drop 掉,将 Hive 表 rename 回去。

  • 使用 CTAS 语句,Spark 示例如下:

CREATE TABLE db.sample_iceberg
(id bigint, ..., dt string) 

USING Iceberg

PARTITIONED BY dt

LOCATION 'qbfs://....'  

TBLPROPERTIES('write.target-file-size-bytes' = '512m', ...)

AS SELECT * FROM db.sample;

在写入完成后进行对数,符合要求后,通过重命名完成切换。

ALTER TABLE db.sample RENAME TO db.sample_backup;

ALTER TABLE db.sample_iceberg RENAME TO db.sample;

CTAS 相比于 migrate 优势是存量数据重新写入,因而可以优化分区、列排序、文件格式和小文件等。缺点是如果存量数据较多,重写耗时耗资源。

以上两个方案,具有如下特性:

优点:

  • 方案简单,执行已有的 SQL 即可

  • 可回滚,原 Hive 表还在

缺点:

  • 写入/读取程序未验证:切换到 Iceberg 表后,可能出现写入或查询异常

  • 要求切换过程停写,对一些业务是不能接受的


3. 爱奇艺****平滑迁移方案

考虑到上述方案的缺点,我们设计了原地双写 + 透明切换的方案,实现平滑迁移,如图 2-1 所示:

  • 建表**:**创建与 Hive 相同 schema 的 Iceberg 表,同步 Hive 表的 TTL、权限等元信息到 Iceberg 表。

  • 历史数据迁移到 Iceberg**:**Hive 历史数据通过 add_file procedure 添加到 Iceberg 中,该操作会根据 Hive 数据构建出 Iceberg 的元数据,实际上 Iceberg 的元数据中指向的是 Hive 的数据文件,减少了数据冗余及历史数据同步时间。

  • 增量数据双写**:**通过爱奇艺自研的 Pilot SQL 网关探测 Hive 表的写入任务,自动复制写入 SQL,并将输出替换成 Iceberg 表,实现双写。

  • **数据一致性****校验:**当历史数据同步完成且增量双写到一定次数之后,后台会自动发起对数,校验 Hive 和 Iceberg 中的数据是否一致。对于历史数据与增量数据会选取一部分数据进行 count 以及字段 CRC 数值校验。

  • 切换**:**数据一致性校验完成后,进行 Hive 和 Iceberg 的切换,用户不需要修改任务,直接使用原来的表名进行访问即可。正常切换过程耗时在几分钟之内。

图 2-1 Hive 切换到 Iceberg 大致流程

图 2-2 展示了 Hive to Iceberg 相关操作界面,点击创建转化任务即可开始进行切换流程,当任务创建成功会在下方展示任务的状态以及运行阶段等信息。

图 2-2 Hive to Iceberg 相关操作界面

03

   核心收益 - 加速查询

1. Iceberg 查询加速技术

Iceberg 自身提供了三层数据过滤策略,分别是 [2]:

分区剪裁:和 Hive 表类似,对于分区表,引擎端可以自动从 where 条件中根据分区键直接提取出需要访问的分区,从而避免扫描所有的分区。分区剪裁可以细分为静态分区剪裁和动态分区剪裁,其中静态分区剪裁发生在 SQL 语句编译阶段,而动态分区剪裁则发生在 SQL 语句执行阶段。

文件过滤:Iceberg 提供了文件级别的统计信息,例如 Min/Max 等,可以快速过滤无关数据和文件,可以用 where 语句中的过滤条件去判断目标数据是否存在于文件中。例如 SELECT * FROM table WHERE dt='2023-01-01' AND channel_id = '20',dt 是分区,channel_id 是字段,对于 channel_id = '20' 这样的过滤条件,元信息中存储了每个文件 channel_id 的 upper_bounds 和 lower_bounds,可以通过判断列值是否在范围内决定是否需要扫描当前文件。

但实际使用中,这种过滤发挥的作用比较小。因为数据写入是随机且无序的,导致 upper_bounds 和 lower_bounds 范围重合度非常高,这种情况下目标数据可能会分布在大部分文件甚至所有文件,扫描数据文件的范围也大大增加。因此在切换为 Iceberg 后,我们可以基于过滤条件中的高频列进行排序,降低文件级别的 upper_bounds 和 lower_bounds 的范围重合度。

除了 MinMax 外,Iceberg 还可以支持更多类型的索引进行文件级过滤,例如字典、布隆过滤器等。

文件内 RowGroup 过滤:对于 Parquet、ORC 这类列式存储文件格式,在文件内部也存在相应的统计信息,例如Min、Max、BloomFiter 等等,利用这些信息可以快速跳过无关的 RowGroup 或者 Stripe,减少文件内数据扫描的量。

2. Iceberg 加速技巧

基于 Iceberg 查询更快的基本原理,我们可以总结出如下技巧:

  • 配置分区:使用分区剪裁的方式使查询只针对特定分区的数据执行,而不需要扫描整个数据集。

  • 指定排序列:通过对数据分布进行合理的组织,最大限度的发挥文件级别的过滤效果,使得查询只集中在特定的文件。例如通过下面的方式使得写入 sample 表的数据按照 category, id 降序写入,注意由于多了一个排序的环节,这种方式会比非排序的写入耗时长。

ALTER TABLE db.sample WRITE ORDERED BY category, id DESC

  • 高基数列应用布隆过滤器:在查询数据时,会自动应用布隆过滤器来快速验证查询数据是否存在于某个数据块,避免不必要的磁盘访问。

write.parquet.bloom-filter-enabled.column.test = true -- parquet 文件给 test 列增加 bloom-filter

write.orc.bloom.filter.columns = test -- orc 文件给 test 列增加 bloom-filter

  • 使用 Trino 代替 Spark:由于 Trino 自身 MPP 的架构,在查询上相较于 Spark 更有优势,并且 Trino 自身对 Iceberg 也有相应的优化,因此如果有秒级查询的需求,可将引擎由 Spark 切换到 Trino。

  • Alluxio 缓存:使用 Alluxio 作为数据缓存层,将数据缓存在内存中。在查询时可以直接从内存中获取数据,避免从磁盘读取数据的开销,可大大提高查询速度,也可防止 HDFS 抖动对任务的影响。

  • ORC 代替 Parquet:由于 Trino 对 ORC 格式有特定的优化,使得 ORC 的读取性能要优于 Parquet,可以将文件格式设置为 ORC 加速查询。

  • 配置合并:写 Iceberg 的任务往往会出现写入文件较小但数量较多的情况,通过将小文件合并成一个或少量更大的文件,有利于减少读取的文件数,降低磁盘 I/O。


3. 性能评测

3.1 文件内过滤性能提升

背景:数据集市是从 Hive 表切换为 Iceberg 表的场景之一,在切换到 Iceberg 后查询速度明显地变快。经过实验对比,确认性能是由文件内 RowGroup 过滤带来的。

图 3-1 Hive 和 Iceberg 查询对比

3.2 列排序对文件内过滤性能提升

我们在另一个场景进一步探索排序对性能的影响。由于分区下仅一个文件,因而文件级过滤不起作用。我们分别比较了 Parquet 和 ORC 这两种文件格式,在排序和未排序下的查询性能,最终结论如下:

  • 同样的文件格式,排序后文件内过滤效果更好,大致能快 40%;

  • ORC 查询性能优于 Parquet;

  • 使用 Trino 查询,我们推荐 Iceberg 表 + ORC 文件格式 + 列排序;

图 3-2 Iceberg 分别在 Parquet、ORC 格式上文件内过滤性能对比

3.3 列排序对文件级过滤性能提升

业务表特定的列可能会频繁用做过滤条件,默认情况下数据是乱序组织,此时列 MinMax 值过滤也难以发挥作用,因特定值在每个文件都被包含。如果在数据写入时,按照该列进行排序组织,则 MinMax 值就能过滤掉大部分无关文件,大幅减少读取的数据量,加速查询。

下面以 CDN 的一个表为例,它的查询频繁用到 isp 和 prov 两个列,一个典型查询如下:

SELECT 

  "date" / 300 * 300 as "date",

  isp, 

  ip_type, 

  sum("traffic") as "traffic"

FROM

  table

WHERE "date" >= 1698986100 AND "date" < 1698986400 

  AND isp IN ('TV', 'Mobile', 'Phone') 

  AND prov IN ('BeiJing') 

 GROUP BY  

  "date" / 300 * 300, isp, ip_type;

我们分别测试对应表,默认不排序,按照 isp 排序,按照 prov 排序 3 种情况,最终性能如下:

  • 按照 prov 排序查询读取数据量是不排序的 25%,耗时是 66%;

  • 按照 isp 排序提升不明显,这是因为 isp 数据量有明显的倾斜,条件中 isp 值占比高达 90%;

图 3-3 Iceberg 文件级过滤性能提升对比

3.4 布隆过滤器的性能提升

在会员订单场景,业务既有基于订单 ID 检索的需求,又有查询某个用户 UserId 历史订单的需求。这两个列基数都非常大,无论用哪个列排序,另一个列的查询都会退化为全表检索。此类场景可以通过布隆过滤器满足。下图演示了开启布隆过滤器后,订单表的性能和 Impala + Kudu 接近,而未开启的情况下查询要接近 1000 秒。

图 3-4 Iceberg 使用布隆过滤器和 Impala + Kudu 的性能对比

3.5 Spark Trino 性能比较

Trino 社区早期版本仅支持 Iceberg 表 V1 的查询,而对 V2 表格式的支持有问题(查询结果不正确)。爱奇艺的方案是在 Pilot SQL 网关中基于 Iceberg 表格式进行路由,V1 表路由到 Trino 引擎,V2 表路由到 Spark 引擎。

我们在 Trino 434 版本,重新验证了 Trino 对 Iceberg V2 表的查询。实验过程如下,对于 TPC-DS 测试集,我们每次变更表 0.1% 的数据,累计变更 20 轮,表使用 Merge On Read 模式,通过 Spark 执行变更生成 Position Delete 文件。changeN 代表 N 次变更后,rewrite_position 代表执行了 Spark rewrite_position_delete_files 后,rewrite_data 代表rewrite_data_files 后。

图 3-5 Spark 和 Trino 对于 Iceberg 的查询性能对比

可以看到:

  • Trino 对于 V2 表查询结果与 Spark 一致,且在相同核数性能优于 Spark,耗时是 Spark 的 1/3 左右;

  • 随着变更轮次的增加(Data File 和 Postition Delete File 数量增加),Trino 查询性能也会逐渐变慢,需要定期进行合并。

===

04

   核心收****益 - 支持变更

1. 变更在业务使用场景

传统上大数据表对变更的支持较差,然而业务上有很多的变更需求:

  • ETL 计算:如广告计费,通过接入 Iceberg 实现变更,简化业务逻辑,实现了更长时间范围的转化回收;

  • 数据修正:批量修正,如对某个数据的状态进行修改、批量删除等;

  • 隐私相关:如播放记录、搜索记录,用户需要删除历史条目等;

  • CDC 同步:如订单业务,需要将 MySQL 中的数据进行大数据分析,通过 Flink CDC 技术很方便地将 MySQL 数据入湖,实时性可达到分钟级。


2. Hive 如何实现变更

在 Hive 中实现变更,主要有如下两种方式:

  • 分区覆写 例如修改某个 id 的相关内容,先筛选出要修改的目标行,更新后与历史数据进行合并,最后覆盖原表。这种方式对不需要修改的数据进行了重写,浪费计算资源;且覆写的粒度最小是分区级别,数据无法进一步细分,任务耗时相对较长。

  • 标记删除 通常的做法是添加标志位,数据初始写入时标志位置 0,需要删除时,插入相同的数据,且标志位置 1,查询时过滤掉标志位为 1 的数据即可。这种方式在语义上未实现真正的删除,历史数据仍然保存在 Hive 中,浪费空间,而且查询语句较为复杂。

3. Iceberg 支持的变更类型

Iceberg 目前支持的变更类型如下:

  • Delete:删除符合指定条件的数据,例如

DELETE FROM table_name WHERE channel_id= '2'

  • Update:更新指定范围的数据,例如

UPDATE table_name SET category='c2' WHERE id='2'

  • MERGE:若数据已存在 UPDATE,不存在执行 INSERT,例如

MERGE INTO db.target t   -- a target table

USING (SELECT ...) s   -- the source updates

ON t.id = s.id  -- condition to find updates for target rows

WHEN MATCHED AND t.count IS NULL AND s.op = 'increment' THEN UPDATE SET t.count = 0 

WHEN NOT MATCHED THEN INSERT *


4. Iceberg 变更策略

Iceberg 支持多种变更策略,每个策略有各自的优劣和适用场景,下面简单介绍一下每种策略的原理 [3]。

  • Copy on Write(写时合并):当进行删除或更新特定行时,包含这些行的数据文件将被重写。写入耗时取决于重写的数据文件数量,频繁变更会面临写放大问题。如果更新数据分布在大量不同的文件,那么更新的执行速度比较慢。这种方式由于结果文件数较少,读取的速度会比较快,适合频繁读取、低频批次更新的场景。

  • Merge on Read(读时合并):文件不会被重写,而是将更改写入新文件,当读取数据时,将新文件合并到原始数据文件得到最终结果。这使得写入速度更快,但读取数据时必须完成更多工作。写入新文件有两种方式,分别是记录删除某个文件对应的行(position delete)、记录删除的数据(equality detete)。

  • Position Delete:当前 Spark 的实现方式,记录变更对应的文件及行位置。这种方式不需要重写整个数据文件,只需找到对应数据的文件位置并记录,减少了写入的延迟,读取时合并的代价较小。

  • Equality Delete:当前 Flink 的实现方式,记录了删除数据行的主键。这种方式要求表必须有唯一的主键,写入过程无需查询数据文件,延迟最低;然而它的读取代价最大,这是由于读取时需要将 equality delete 记录和所有的原始文件进行 JOIN。

表 4-1 总结了不同变更策略的特点及适用场景:

表 4-1 Iceberg 不同变更策略对比

Iceberg **配置变更策略:**Iceberg 中可以通过 write.delete.mode、write.merge.mode、write.update.mode 属性分别设置删除、合并、更新等写入模式,默认值均是 copy-on-write。当前只有 V2 表支持 Merge-on-read 模式。

表 4-2 Iceberg 变更属性配置方式

5. 业务接入

本节通过一些例子,说明 Iceberg 支持变更给业务带来的价值。

5.1 广告计费转换

如图 4-1 所示,在效果广告场景中,客户有查询计费转化数和深度转化数据的需求(基于计费时间)。比如某垂直领域客户,希望把用户行为统一起来,1 号发生的 100 万曝光,产生了 40 万点击(仅为示例,非真实数据),进而在后续的第 N 天内发生了 5000 的用户付费行为,需将第 N 天的付费归因到  1  号的曝光。广告报表都是基于用户行为时间,即日志时间聚合而成,为支持将深度转换归因到广告计费的当日,由于 Hive 不支持变更做了如下复杂的设计:

  • 每天触发一次计算,从行为表聚合出过去 7 天的“计费时间”数据。此处用 rt 字段代表计费时间

  • 提供统一视图合并行为数据和计费时间数据,计费归因表 rt as dt 作为分区过滤查询条件,满足同时检索曝光和计费转化的需求

图 4-1 广告计费转换场景

而在 Iceberg 场景下,其支持变更因而无需使用多个不同的表,直接在原表通过如下 SQL 即可完成:

MERGE INTO iceberg_taget_table t 

USING (   

  SELECT * FROM changes_table   

  WHERE dt='2023-12-12' 

) s   

ON t.id = s.id 

... 

AND t.dt = s.dt 

WHEN MATCHED THEN 

 UPDATE SET 

  count = count + s.cnt,  

  deep_count = deep_count + s.deep_cnt, 

  ...

通过 Iceberg 表 merge 可简化整个处理流程:

  • 时效性提升:从天级缩短到小时级,客户更实时观察成本,有利于预算引入;

  • 计算更长周期数据:原先为计算效率仅提供 7 日内转换,而真实场景转换周期可能超过 1 个月;

  • 表语义清晰:多表联合变为单表查询。

5.2 数据修正

举个例子,业务发现线上 ETL 任务逻辑有 BUG,导致某个列的值不准确。虽然线上 ETL 任务已经修复,但是错误的数据已经写入到下游的 Iceberg 表里。如果是 Hive 场景,需要重跑 ETL 任务,全量覆盖天分区进行修正。而在 Iceberg 表我们可以通过如下 SQL 进行修正:

UPDATE your_iceberg_table 

SET strategy_code = 'correct_value' 

WHERE dt = '2023-12-01' and strategy_code = 'wrong_value'

05

   总结

Iceberg 不仅提供了数据删除、更新等功能,有效满足数据保留政策和合规性要求,而且查询加速措施更为多样,可以利用列式存储、索引和元数据统计信息来优化查询计划,提高查询性能,可以帮助我们简化业务逻辑,提升时效性,加快数据产出。

通过以上平滑切换方案,从 Hive 到 Iceberg 不需要重新构建数据湖,在尽量保持语义兼容性的情况下,可以进行无缝迁移,减少了迁移的复杂性和风险。

后续我们将继续推进 Hive 到 Iceberg 的迁移,提升数据流通效率,促进业务提效增收。

06

   引用

  1. From Hive Tables to Iceberg Tables: Hassle-Free

  2. 通过数据组织优化加速基于Apache Iceberg的大规模数据分析

  3. Row-Level Changes on the Lakehouse: Copy-On-Write vs. Merge-On-Read in Apache Iceberg

  4. 《爱奇艺数据湖实战 - 综述》

  5. 《爱奇艺数据湖实战 - 广告》

  6. 《爱奇艺数据湖实战 - 基于数据湖的日志平台架构演进》

  7. 《爱奇艺数据湖实战 - 数据湖技术在爱奇艺BI场景的应用》

  8. 《爱奇艺在Iceberg落地相关性能优化与实践》

也许你还想看

数据湖在爱奇艺数据中台的应用

爱奇艺数据湖实战 - 基于数据湖的日志平台架构演进

爱奇艺数据湖实战

相关推荐
关注或联系我们
添加百川云公众号,移动管理云安全产品
咨询热线:
4000-327-707
百川公众号
百川公众号
百川云客服
百川云客服

Copyright ©2024 北京长亭科技有限公司
icon
京ICP备 2024055124号-2