本文为您介绍基于HDFS如何使用Spark、Flink操作Iceberg表。

Spark操作Iceberg

目前，Spark默认部署在/usr/local/路径下，在Spark组件的jars文件夹中需要包含Iceberg的相关jar包，包括iceberg-spark-runtime-xxx.jar、iceberg-spark-extensions-xxx.jar、iceberg-mr-xxx.jar,其中xxx为具体的版本号。例如，在Spark 3.4版本中使用Iceberg 1.6.1版本，其中一个jar包名称为iceberg-spark-runtime-3.4_2.12-1.6.1-1.0.0.jar，jar包中2.12表示Scala版本号、1.0.0表示自研版本号。

1、创建catalog

Iceberg提供catalog功能，您可以使用SQL命令来管理表，并按名称来加载这些表。catalog的配置是通过在spark.sql.catalog.(catalog_name)下的属性来实现的。

您可以在spark-defaults.conf中配置Iceberg的参数，也可以通过spark作业命令配置Iceberg的参数。

Iceberg目前可以使用Hadoop或者Hive类型的catalog。

当您想要在spark-defaults.conf中配置Iceberg的Hadoop类型catalog时，可以按照如下方式配置。该命令表示Spark可以访问保存在HDFS中路径为hdfs://namenodeIp:8020/warehouse/path的Iceberg表数据，其中catalog_name为自定义值。

spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
spark.sql.catalog.catalog_name=org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.catalog_name.type=hadoop
spark.sql.catalog.catalog_name.warehouse=hdfs://namenodeIp:8020/warehouse/path

当您想要在spark-defaults.conf中配置Iceberg的Hive类型catalog时，可以按照如下方式配置。该命令表示Spark使用该catalog会从Hive元数据仓库中加载表数据。其中Host和Port分别表示Hive metastore URL和端口，catalog_name通常设置为spark_catalog。

spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
spark.sql.catalog.catalog_name=org.apache.iceberg.spark.SparkSessionCatalog
spark.sql.catalog.catalog_name.type=hive
spark.sql.catalog.catalog_name.uri=thrift://Host:Port

2、创建表

Spark可以执行如下SQL语句创建一个指定分区的Iceberg表。

CREATE TABLE catalog_name.db_name.table_name (id bigint, name string, category string) USING iceberg PARTITIONED BY (category);
PARTITIONED BY子句支持表达式转换，以创建隐藏分区。
CREATE TABLE catalog_name.db_name.table_name (id bigint, name string, category string, ts timestamp) USING iceberg PARTITIONED BY (bucket(32, id), day(ts), category);

表达式还支持如下类型：

• year(ts)：按年份进行分区。

• month(ts)：按月份进行分区处理。

• day(ts)或date(ts)：相当于dateint分区操作。

• hour(ts)或date_hour(ts)：相当于dateint和按小时进行分区处理。

• bucket(N, col)：根据哈希值除以N后的余数来划分buckets。

• truncate(L, col)：按值进行分区，结果截断为L位。

3、写入数据

按照上述示例，创建一个包含隐藏分区的Iceberg表后，使用INSERT INTO语法执行数据的写入操作。

INSERT INTO catalog_name.db_name.table_name(id,name,category,ts) VALUES(1, 'john', 't1', TIMESTAMP'2026-08-17 09:30:00'), (2, 'Tony', 't2', TIMESTAMP'2026-08-16 09:30:00'), (3, 'lady', 't3', TIMESTAMP'2026-08-19 09:30:00');

4、读取数据

在往Iceberg表中写入一定数量的数据后，使用SELECT查询语法执行数据的读取操作。

SELECT * from catalog_name.db_name.table_name;

Flink操作Iceberg

Iceberg支持Flink的DataStream API和Table API。在Flink中创建Iceberg表时，建议使用Flink SQL客户端，因为这样使用更加便捷。

在/usr/local路径下部署完Flink后，需要先启动一个独立的Flink集群。

/usr/local/flink/bin/start-cluster.sh

检测Flink的lib文件夹，确保存在iceberg-flink-runtime-xxx.jar。然后执行如下命令，启动Flink SQL客户端。

/usr/local/flink/bin/sql-client.sh

然后输入如下参数，使得Flink SQL 作业可以提交到Yarn上执行。

SET 'execution.target' = 'yarn-per-job';
SET execution.checkpointing.interval = 30s;

1、创建catalog

Flink支持使用Flink SQL创建catalog,可以创建Hive类型的catalog。

参考如下示例，注册一个名为hive_test的Hive类型catalog，其中Hive的部署路径为/usr/local/hive，因此示例中'hive-conf-dir'值为/usr/local/hive/conf。

CREATE CATALOG hive_test WITH (
'type'='iceberg',
'catalog-type'='hive',
'hive-conf-dir'='/usr/local/hive/conf'
);

2、创建表

参考如下示例，基于hive_test catalog，创建一个名为test的Iceberg分区表。

CREATE TABLE `hive_test`.`default`.`test` (id bigint, name string, category string) PARTITIONED BY (category);

3、写入数据

在Iceberg表被创建后，使用INSERT INTO语法执行数据的写入操作。

INSERT INTO `hive_test`.`default`.`test` VALUES (1, 'john', 't1'), (2, 'Tony', 't2'), (3, 'lady', 't3');

4、读取数据

在往Iceberg表中写入一定数量的数据后，使用SELECT查询语法执行数据的读取操作。

SELECT * from `hive_test`.`default`.`test`;