# FlinkServer对接Iceberg
#### 操作场景
本章节提供了如何使用FlinkServer写FlinkSQL对接Iceberg的操作指导。Iceberg可以作为Sink表和Source表。
#### 约束与限制
- 本章节适用于MRS 3.6.0-LTS.1及之后的版本。
- Flink On Iceberg不支持JDK 8运行。
- 通过FlinkServer提交Flink On Iceberg的Flink Jar类型作业，需要设置为JDK 21 ，具体操作请参考[FlinkServer提交的作业，JDK版本切换操作步骤](https://support.huaweicloud.com/cmpntguide-lts-mrs/mrs_01_300891.html#mrs_01_300891__zh-cn_topic_0000002418300824_section843122313405)。
- 通过Flink客户端提交Flink On Iceberg的作业，需要设置为JDK 21，具体操作请参考[Flink客户端提交的作业，JDK版本切换操作步骤](https://support.huaweicloud.com/cmpntguide-lts-mrs/mrs_01_300891.html#mrs_01_300891__zh-cn_topic_0000002418300824_section9347187192215)。
 
#### 前提条件
- 集群中已安装HDFS、ZooKeeper、Yarn、Flink、Hive组件。
- 参考[创建FlinkServer权限角色](https://support.huaweicloud.com/cmpntguide-lts-mrs/mrs_01_24049.html)创建一个具有FlinkServer管理员权限的用户用于访问Flink WebUI，如：**flink_admin**。
- 参考[FlinkServer Catalog管理](https://support.huaweicloud.com/cmpntguide-lts-mrs/mrs_01_300802.html#mrs_01_300802)创建一个Iceberg Catalog，如：iceberg_catalog。
- 仅支持Flink与Hive组件共集群，不支持Hive多服务。
 
#### Iceberg作为Sink表
1. 使用**flink_admin** 登录Manager界面，选择"集群 \> 服务 \> Flink"，在"Flink WebUI"右侧，单击链接，访问Flink的WebUI。
   
   登录集群Manager具体操作，请参考[访问MRS集群Manager](https://support.huaweicloud.com/usermanual-mrs/mrs_01_0128.html)。
   
   
2. 新建一个Iceberg表，例如：iceberg_catalog.\`default\`.iceberg_table，具体操作可参考[FlinkServer数据管理](https://support.huaweicloud.com/cmpntguide-lts-mrs/mrs_01_300803.html)，完整示例如下：
   
   ```
   CREATE TABLE if not exists iceberg_catalog.`default`.iceberg_table (
     id BIGINT COMMENT 'unique id',
     data STRING NOT NULL
   ) PARTITIONED BY (data)
   WITH
     ('format-version' = '2');
   ```
   
   
3. 参考[新建作业](https://support.huaweicloud.com/cmpntguide-lts-mrs/mrs_01_24024.html#mrs_01_24024__section1746418521537)，新建Flink SQL流作业，参考如下内容在作业开发界面进行作业开发，根据业务要求配置"checkpoint"间隔，配置完成后启动作业。
   
   ```
   CREATE TABLE datagen(id BIGINT, data STRING) WITH (
     'connector' = 'datagen',
     'rows-per-second' = '5'
   );
   insert into
     iceberg_catalog.`default`.iceberg_table
   select
     *
   from
     datagen;
   ```
   
   
 
#### Iceberg作为Source表
1. 使用**flink_admin**登录Manager界面，选择"集群 \> 服务 \> Flink"，在"Flink WebUI"右侧，单击链接，访问Flink的WebUI。
2. 参考[新建作业](https://support.huaweicloud.com/cmpntguide-lts-mrs/mrs_01_24024.html#mrs_01_24024__section1746418521537)，新建Flink SQL流作业，参考如下内容在作业开发界面进行作业开发，配置完成后启动作业。
   
   ```
   CREATE TABLE print(id BIGINT, data STRING) WITH ('connector' = 'print');
   insert into
     print
   select
     *
   from
     iceberg_catalog.`default`.iceberg_table
     /*+ OPTIONS('streaming'='true', 'monitor-interval'='60s')*/ ;
   ```