# FlinkServer作业对接Iceberg
#### 操作场景
本章节提供了如何使用FlinkServer写FlinkSQL对接Iceberg的操作指导。Iceberg可以作为Sink表和Source表。
本章节仅适用于MRS 3.6.0-LTS.1及之后版本。
#### 前提条件
- 集群中已安装HDFS、Zookeeper、Yarn、Flink、Hive组件。
- 参考[创建FlinkServer权限角色](https://support.huaweicloud.com/cmpntguide-lts-mrs/mrs_01_24049.html)创建一个具有FlinkServer管理员权限的用户用于访问Flink WebUI，如：**flink_admin**。
- 仅支持Flink与Hive组件共集群，不支持Hive多服务。
 
#### Iceberg作为Sink表
1. 使用**flink_admin** 登录Manager界面，选择"集群 \> 服务 \> Flink"，在"Flink Web UI"右侧，单击链接，访问Flink的WebUI。
   
   登录集群Manager具体操作，请参考[访问MRS集群Manager](https://support.huaweicloud.com/usermanual-mrs/mrs_01_0128.html)。
   
   
2. 参考[集群连接模式创建Flink SQL作业](https://support.huaweicloud.com/cmpntguide-lts-mrs/mrs_01_24024.html)，新建Flink SQL流作业，参考如下内容在作业开发界面进行作业开发，配置完成后启动作业。
   
   需勾选"基础参数"中的"开启CheckPoint"，"时间间隔（ms）"可设置为"60000"，"模式"可使用默认值。
   建议开启故障恢复策略，提高作业可靠性。例如"故障恢复策略"选择"fixed-delay"，"重试次数"设置为"3"，"失败重试间隔"设置为"30"，重试次数和间隔可按实际业务需要填写。
   作业启动，状态显示"运行中"后，可通过"更多 \> 作业详情"跳转到Flink作业的原生UI页面，查看Job运行情况。
   ```
   CREATE TEMPORARY TABLE datagen(id BIGINT, data STRING) WITH (
     'connector' = 'datagen',
     'rows-per-second' = '5'
   );
   CREATE TABLE flink_table (id BIGINT, data STRING) WITH (
     'connector' = 'iceberg',
     'catalog-name' = 'hive_prod',
     'catalog-database' = 'default',
     'catalog-table' = 'hive_iceberg_table',
     'uri' =  'Hive客户端“Hive/config/hive-site.xml”文件中“hive.metastore.uris”的值',
     'hive.metastore.kerberos.principal' = 'Hive客户端“Hive/config/hive-site.xml”文件中“hive.metastore.kerberos.principal”的值' --普通模式集群不需要该参数
   );
   insert into
     flink_table
   select
     *
   from
     datagen;
   ```
   
   
 
#### Iceberg作为Source表
1. 使用**flink_admin** 登录Manager界面，选择"集群 \> 服务 \> Flink"，在"Flink Web UI"右侧，单击链接，访问Flink的WebUI。
   
   登录集群Manager具体操作，请参考[访问MRS集群Manager](https://support.huaweicloud.com/usermanual-mrs/mrs_01_0128.html)。
   
   
2. 参考[集群连接模式创建Flink SQL作业](https://support.huaweicloud.com/cmpntguide-lts-mrs/mrs_01_24024.html)，新建Flink SQL流作业，参考如下内容在作业开发界面进行作业开发，配置完成后启动作业。
   
   需勾选"基础参数"中的"开启CheckPoint"，"时间间隔（ms）"可设置为"60000"，"模式"可使用默认值。
   建议开启故障恢复策略，提高作业可靠性。例如"故障恢复策略"选择"fixed-delay"，"重试次数"设置为"3"，"失败重试间隔"设置为"30"，重试次数和间隔可按实际业务需要填写。
   作业启动，状态显示"运行中"后，可通过"更多\>作业详情"跳转到Flink作业的原生UI页面，查看Job运行情况。
   ```
   CREATE TEMPORARY TABLE print(id BIGINT, data STRING) WITH ('connector' = 'print');
   CREATE TABLE flink_table (id BIGINT, data STRING) WITH (
     'connector' = 'iceberg',
     'catalog-name' = 'hive_prod',
     'catalog-database' = 'default',
     'catalog-table' = 'hive_iceberg_table',
     'uri' = 'Hive客户端“Hive/config/hive-site.xml”文件中“hive.metastore.uris”的值',
     'hive.metastore.kerberos.principal' = 'Hive客户端“Hive/config/hive-site.xml”文件中“hive.metastore.kerberos.principal”的值' --普通模式集群不需要该参数
   );
   insert into
     print
   select
     *
   from
     flink_table
     /*+ OPTIONS('streaming'='true', 'monitor-interval'='60s')*/ ;
   ```