Amoro Mixed Format 在上海钢联的构建实时湖仓实践

Amoro 是一个构建在 Apache Iceberg 等开放数据湖表格之上的湖仓管理系统，提供了一套可插拔的数据自优化机制和管理服务，旨在为用户带来开箱即用的湖仓使用体验。

01 作者简介

熊俊，钢联大数据开发工程师，目前主要参与 Flink 实时计算和基于Apache lceberg 的湖仓一体方向的研发。

王涛，网易资深平台开发工程师，主要从事大数据和湖仓平台建设。

02 业务背景

上海钢联是一家全球领先的大宗商品及相关产业数据服务商，积累了包括黑色金属、有色金属、能源化工、农产品等行业、上下游数百条产业链的海量数据。

在实际业务场景中，有很多对数据实时性要求较高的报表、质检等业务需求，这些业务数据既有实时的埋点数据，也有来源于数据库的 CDC 数据，需要下游的数仓能提供高效的更新能力。在实践中，我们引入了较多的技术栈来解决不同场景的问题，包括 Hive、Kafka、HBase、Kudu、ClickHouse 等，由于缺少统一的方案，既增大了维护成本，也造成了数据在多个系统中重复保存。

在这样的背景下，我们产生了架构层面改造的诉求，希望借助数据湖技术来构建流批一体的湖仓系统，以达到统一数据管理、减少技术栈、节省资源的目的。

03 引入 Apache Iceberg

我们尝试引入了 Iceberg，Iceberg 提供的 upsert 和分钟级延迟的能力，以及对 Flink、Spark、Trino 等多种引擎的适配，能契合业务在数据实效性、引擎选型方面的需求。但是，在落地 Iceberg 的过程中，我们也遇到了几个痛点：

1.小文件问题

Iceberg 在频繁提交（每分钟提交一次）时，会产生很多小文件，为了解决这个问题，我们开发了一套小文件合并程序，周期性调度执行合并功能，然而一旦出现合并不及时的情况，后续执行合并时经常出现 OOM、提交冲突等问题，影响了服务的稳定性，也造成了一些提交频繁、流量比较大的表，落地困难。

2.upsert 场景下无法增量消费 Iceberg 表

upsert 需求在我们的业务场景下是非常普遍的，不只是来自数据库的 CDC 数据，业务的埋点数据由于可能存在重放的情况，也需要支持upsert，否则会造成下游数据重复。Iceberg 虽然支持 upsert，但是在开启了 upsert 之后，增量消费 Iceberg 表的数据还不支持，给我们数据的分层建设造成了困难。

这些问题限制了我们线上使用 Iceberg 的规模，通过调研和测试，我们发现，Amoro 在 Iceberg 的基础上，提供了小文件解决方案以及增强的 upsert 功能，可以解决我们使用 Iceberg 中的痛点。

04 引入 Amoro Mixed Iceberg

Amoro Mixed Iceberg 表基于我们比较熟悉的 Iceberg 表构建，在保留了 Iceberg 表的特点的同时，还提供了一些增强。更多关于 Mixed Format 原理介绍，请参考公众号文章：Amoro Mixed Format 适用场景及原理解析。

使用 Amoro Mixed Iceberg，给我们带来这些价值：

1.提供增强的 upsert 能力

借助 Amoro Mixed Iceberg 有主键表，既支持了 ODS 层的 upsert，也支持增量消费 CDC 数据，继续构建下游的表；尤其是还可以通过 Log Store 将数据写入 Kafka，给下游提供毫秒级延迟的数据，避免数据流经多个层级后延迟不断叠加。

2.开箱即用的数据优化能力

AMS 服务提供了开箱即用的自动合并文件、过期文件、孤儿文件清理等能力，这些功能在之前使用 Iceberg 时都需要通过定时任务执行；尤其是持续自动合并文件的能力，解决了之前定期调度批任务出现 OOM 的问题；在 Amoro 页面上启动和停止合并任务，也方便了我们对合并资源的管理和维护。

3.提升数据时效性

借助 MOR（Merge-On-Read）能力，通过 Trino 引擎可以查询到延迟为分钟级别的数据，可以满足实时报表、质检的需求；另外借助持续的文件合并，可以将 MOR 查询耗时控制在合理的水平，不会因为小文件持续增加，对查询性能产生影响。

05 落地方案

概述

如上图所示，在存储层，以 Amoro Mixed Iceberg 表为基础，我们构建起了流批一体的湖仓系统，逐步替代之前由 Hive 表构建离线链路，Kafka 构建实时链路的用法。

其中，ODS 层全部由 Amoro 表构建，承接实时写入的数据库和日志数据；EDW 层也是基于 Amoro 表通过 Flink 任务的双流 join、维表 join 构建；ADS 层我们还是保留了 ClickHouse、ES、数据库等多种组件，通过流和批两种方式写入。

在计算层，绝大部分计算是基于 Flink SQL 完成的，同时，我们还引入了 Trino，给业务提供了分钟级的 Ad-Hoc 查询的能力。

目前，我们已经接入了30+ 的 Mixed Iceberg 表，支持2000万/天的更新流量。

表的管理Amoro 提供了开箱即用的表管理能力，表的元数据保存在 AMS 中，能够方便的完成表的创建和维护。

新建 Amoro 表在选型时，考虑到业务上数据去重的需求，我们全部选择了有主键表，并且开启了 upsert 能力。

表的维护做到了开箱即用，只需要通过少量的配置，就可以完成表的合并文件、数据过期等需求。目前，我们使用的是 Amoro 社区最新 release 的 0.5.0 版本，并部署了 Flink Optimizer 进行文件合并，满足了我们生产场景持续进行文件合并的诉求。

数据入湖

数据入湖主要来源于两类数据，一类是数据库的 CDC 数据，另一类是埋点日志数据，这两类数据 通过 Flink 任务写入由 Amoro Mixed Iceberg 表构建的 ODS 层，如下图所示。

数据库 CDC 数据入湖由 Flink CDC 任务完成，为了减小对数据库的压力，我们用到了 Flink CDC 的整库入湖的功能。

日志数据来源于 Kafka，由于上游 Kafka 数据可能出现数据重复写入的情况，我们给每行数据定义了唯一 ID，并将唯一 ID 作为 Amoro Mixed Iceberg 表的主键，借助 Mixed Iceberg 表的 upsert 能力，在 ODS 层解决了数据重复的问题。

实时构建下游数据

从 ODS 层构建 EDW 层，以及构建ADS 层的过程中，依赖了 Mixed Iceberg 增量消费 CDC 数据的能力：对于实时性要求较高的表，因为开启了 Log Store，CDC 数据来源于 Kafka，提供了毫秒级的延迟；对于没有开启 Log Store 的表，CDC 数据来源于 Change Store，也能提供大约1～5分钟的延迟。

在构建 EDW 层和 ADS 层时，我们主要用到了两类 Flink 的 join 语义，一类是双流 join，一类是维表 join。

双流 join 主要是消费上游 Amoro 表的增量数据实现，而维表 join场景则主要大部分还是基于自定义连接器 flink-sql-connector-clickhouse 实现 Amoro 增量数据和 ClickHouse维表 lookup join，如下是一个 Amoro 与 ClickHouse 做维表 join 的示例。

CREATE CATALOG dw_amoro_catalog WITH ( 'type' = 'arctic', 'properties.auth.simple.hadoop_username' = 'hdfs', 'properties.auth.type' = 'simple', 'metastore.url'='zookeeper://xxx:xxx/dw_amoro_cluster/dw_amoro_catalog');
create TEMPORARY table xxx ( `id` BIGINT , `name` STRING, `update_time` TIMESTAMP(3), `_sign` TINYINT, PRIMARY KEY (id) NOT ENFORCED) WITH ( 'connector' = 'clickhouse', 'url' = 'clickhouse://xxx:xx', 'username' = 'xx', 'password' = 'xx', 'database-name' = 'ODS', 'table-name' = 'xx', -- LOOK UP JOIN 'lookup.cache.max-rows'='200000', 'lookup.cache.ttl'='100000', -- SINK 'sink.batch-size' = '1000', 'sink.flush-interval' = '5000', 'sink.max-retries' = '3');
-- insert into xxxSELECT FROM_UNIXTIME(A.INCOMETIME / 1000 , 'yyyy-MM') DATA_DATE,A.MEMBERID MEMBER_ID, mm.name MEMBER_NAME ,FROM_UNIXTIME(A.INVOICETIME / 1000 , 'yyyy-MM-dd HH:mm:ss') INVOICETIME, FROM_UNIXTIME(A.INCOMETIME / 1000 , 'yyyy-MM-dd') INCOMETIME, LISTAGG(DISTINCT D.NAME) FNC_TYPE_NAME,A.AMOUNT, A.STATUS, A.INVOICECODE, A.ADMINID ADMIN_ID FROM (select *,PROCTIME() ts from dw_amoro_catalog.ods.oracle_new_finance_fnc_finance) A LEFT JOIN dw_amoro_catalog.ods.xxx B ON A.INVOICEID = B.INVOICEID LEFT JOIN dw_amoro_catalog.ods.xxx C ON B.INVOICEID = C.INVOICEID LEFT JOIN dw_amoro_catalog.ods.xxx D ON C.BUSINESSTYPEID = D.TYPEID JOIN xxx FOR SYSTEM_TIME AS OF A.ts AS mm ON mm.id =A.MEMBERID WHERE A.STATUS = 2 AND FROM_UNIXTIME(A.INCOMETIME / 1000, 'yyyy-MM') = FROM_UNIXTIME(UNIX_TIMESTAMP() , 'yyyy-MM') -- AND A.ADMINID = 220877 AND mm._sign = 1 GROUP BY A.INVOICEID, A.MEMBERID,mm.name, A.INVOICETIME, A.INCOMETIME, A.AMOUNT, A.STATUS, A.INVOICECODE ,A.ADMINID;

MOR 查询

我们部署了 Trino 集群来满足用户 Ad-Hoc 查询的需求，借助于 Amoro 在 Trino 引擎中支持了 Merge on read 的能力，可以给用户提供分钟级延迟的查询结果。目前 Trino 集群的规模不大（5 * 24G），但是也足够满足用户当前的查询需求，查询耗时稳定控制在秒级。

06 未来规划

1. 借助 Flink MOR 读取 Amoro Mixed Format Table 构建下游表

我们的技术栈主要以 Flink 为主，但仍然有通过批任务构建下游表的场景，因此使用 Flink 批模式MOR 读取 Mixed Format 表就是一个很好的实现路径，目前这个功能即将在社区发布，我们将进行测试验证和使用。

2.引入 Mixed Format 的 Look up join 功能

引入 Mixed Format 的 Look up join 能力，在部分维表 join的场景下替换掉 ClickHouse。

3.用 Mixed Format Table替换掉业务对 Kudu、Impala 的使用

规避 Kudu Impala 使用中的数据孤岛、内存占用高等问题，也进一步统一技术栈，减小维护成本。

07 总结

总体而言，通过借助 Amoro 开箱即用的能力，以及 Amoro Mixed Iceberg 在 upsert 和 MOR 等方面的功能特性和优化，我们成功解决了在生产场景中流批一体、数据实效性等方面的需求。未来，我们将继续关注和参与 Amoro 开源社区的发展。

Amoro 最近发布了0.5.0版本，在专门的试用群有 mentor 全程进行一对一辅导，社区还为试用和贡献的小伙伴精心准备了社区礼包和 Air pods。下一个实践案例等你来共建~

END

看到这里记得关注、点赞、转发 一键三连哦~

精彩回顾：

Amoro Mixed Format 适用场景及原理解析

Amoro 0.5.0 版本发布，邀测开启试用有礼

Apache Iceberg + Arctic 构建云原生湖仓实战企查查基于 Apache Iceberg 与 Arctic 构建实时湖仓实践
关于 Amoro 的更多资讯可查看：官网：https://amoro.netease.com/源码：https://github.com/NetEase/amoro社群：后台回复【社群】或扫描下方二维码↓，邀你进群

点击下方【阅读原文】直达 Amoro 官网