文档首页/ MapReduce服务 MRS/ 组件操作指南（LTS版）/ 使用Flink/ 集群连接模式创建FlinkServer作业/ 配置FlinkServer对接其他组件/ FlinkServer作业对接Iceberg
更新时间：2026-06-29 GMT+08:00
查看PDF
分享

FlinkServer作业对接Iceberg

操作场景

本章节提供了如何使用FlinkServer写FlinkSQL对接Iceberg的操作指导。Iceberg可以作为Sink表和Source表。

本章节仅适用于MRS 3.6.0-LTS.1及之后版本。

前提条件

  • 集群中已安装HDFS、Zookeeper、Yarn、Flink、Hive组件。
  • 参考创建FlinkServer权限角色创建一个具有FlinkServer管理员权限的用户用于访问Flink WebUI，如：flink_admin。
  • 仅支持Flink与Hive组件共集群，不支持Hive多服务。

Iceberg作为Sink表

  1. 使用flink_admin登录Manager界面，选择“集群 > 服务 > Flink”，在“Flink Web UI”右侧，单击链接，访问Flink的WebUI。

    登录集群Manager具体操作，请参考访问MRS集群Manager。

  2. 参考集群连接模式创建Flink SQL作业，新建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。

  2. 参考集群连接模式创建Flink SQL作业，新建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')*/ ;

相关文档