组件版本
- Kubernetes 1.21.5
- CloudEon 1.2.0
- Iceberg 1.2.1
- Amoro 0.5.0
- Spark 3.2.3
- Flink 1.15.4
- Hadoop 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实现数据湖表的自动优化性能，为大数据应用提供了可靠的基础设施。
