Iceberg+Amoro+CloudEon 体验云原生数据湖

组件版本

Kubernetes 1.21.5cloudeon 1.2.0Iceberg 1.2.1Amoro 0.5.0Spark 3.2.3Flink 1.15.4Hadoop 3.3.4

在 Kubernetes 上搭建大数据基础环境

之前的文章有介绍过怎么在 Kubernetes 上快速搭建大数据基础环境，这里就不重复介绍了。安装完后，可以看到如下图各个基础服务都启动完成。

安装 Amoro

Amoro 是一个开源的数据湖管理系统。在安装 Amoro 之前，需要选择要安装的服务和节点，并准备好 MySQL 数据库。选择安装服务

分配角色实例到指定节点安装

修改初始化配置，需要填写 mysql 信息，数据库得提前自行准备好

安装成功

可以用在 Kubernetes 集群上看到对应的 pod

开启 Iceberg 支持

Iceberg 是一个数据湖开源项目，用于管理大规模数据的表格式。要在 Cloudeon 上启用 Iceberg 支持，需要配置相应的参数。Spark 开启 Iceberg 插件：

Flink 开启 Iceberg 插件：

HDFS 上创建 Iceberg 数据目录

为了使用 Iceberg，需要在 HDFS 上创建 Iceberg 的数据目录。连接 hdfs-namenode 的 pod shell，在 hdfs 上创建 Iceberg 的数据目录/Iceberg/warehouse

Amoro 上托管 Iceberg

在 Amoro 中，您可以托管 Iceberg 表，实现数据湖表的管理和自优化。打开 Amoro 服务页面，默认密码是 admin/admin

配置 catalog 托管 Iceberg。这里需要上传 cloudeon 安装的 hdfs 服务的 core-site.xml 和 hdfs-site.xml 文件。选择外部 Catalog，metadata 选择 hadoop。需要设置属性 warehouse=/Iceberg/warehouse。

配置 Optimizer Group，先要添加 group

选择 FlinkContainer，需要手动设置参数 taskmanager.memory 和 jobmanager.memory

设置 scale out，配置并行度，点击确定后，Amoro 会启动一个 Flink 作业并提交到 YARN 上

等待一会，可以看到 Optimizers 识别到了 YARN 上的 Flink 作业

也可以看到在 YARN 上看到 Flink 作业正在运行了

Flink 生成流式数据写入 Iceberg

使用 Flink SQL，可以生成流式数据并将其写入 Iceberg 表。连接 Flink-history 的 pod shell，启动 Flink standalone 集群，生成一些测试数据。

standalone 集群启动成功后，启动 sql-client。

sql-client.sh

打开后，输入脚本测试。

CREATE CATALOG iceberg_catalog WITH (  'type'='iceberg',  'catalog-type'='hadoop',  'warehouse'='hdfs:///iceberg/warehouse');
CREATE TABLE datagen_v1 ( id INT, user_id INT, city STRING, cost double, order_time TIMESTAMP) WITH ( 'connector' = 'datagen', 'rows-per-second'='5', 'fields.id.kind'='sequence', 'fields.id.start'='1', 'fields.id.end'='1000', 'fields.cost.min'='1', 'fields.cost.max'='1000', 'fields.city.length'='10');


CREATE TABLE `iceberg_catalog`.`default`.`order` (    id BIGINT COMMENT 'unique id',    user_id BIGINT,    city STRING,    cost DOUBLE,   order_time TIMESTAMP,   PRIMARY KEY (id) NOT ENFORCED) WITH (    'format-version'='2',     'write.upsert.enabled'='true',    'write.metadata.metrics.default'='full');
SET execution.checkpointing.interval = '30s';
insert into `iceberg_catalog`.`default`.`order`  select * from datagen_v1;

这段脚本执行完后，会创建 Iceberg 表，并在 hdfs 上创建存储目录，Flink 会不断插入测试数据到 Iceberg 表中。

Spark 写入数据

Spark 也可以用于写入数据到 Iceberg 表。连接 Spark-thriftserver 的对外暴露的端口，使用 DBeaver 连接后进行操作

set `spark.sql.iceberg.handle-timestamp-without-timezone`=`true` ;

insert into `default`.`order` values(9990,1,'beijing',200,timestamp("2023-09-01 12:31:00")),(9991,2,'shanghai',322,timestamp("2023-09-01 12:32:00")),(9992,3,'beijing',100,timestamp("2023-09-01 12:33:00"));
select * from `default`.`order` WHERE id >9000;
UPDATE   `default`.`order` set cost= cost+1000  WHERE id >9000;

Amoro 自动优化 Iceberg 表

Amoro 能够自动优化 Iceberg 表的性能，包括文件存储优化。可以通过 Amoro 的界面查询 Iceberg 表的 schema 信息和文件信息，以查看优化效果。也可以通过 Web 终端查询 Iceberg 的数据。

打开 Amoro，可以看到 order 表详情。

Amoro 会自动优化表存储的文件

也可以通过 Amoro 的 web 终端进行查询 Iceberg 的表。

set `spark.sql.iceberg.handle-timestamp-without-timezone`=`true` ;select * from `default`.`order`;

Amoro 的 web 终端内置使用的是 Sparksql，用的是 local 模式。Amoro 也支持 web 终端对接 kyuubi，后面可以试试。

总结

通过使用 Cloudeon、Iceberg、Flink、Spark 和 Amoro，您可以轻松地在 Kubernetes 上构建一个强大的大数据基础环境。这个环境不仅支持流式数据生成和处理还有批处理，还能借助 Amoro 实现数据湖表的自动优化性能，为大数据应用提供了可靠的基础设施。