前言：

        在 Flink 和 Iceberg 的集成方面，社区实现了Iceberg 的 Flink Streaming Reader，意味着我们可以通过 Flink 流作业增量地去拉取 Apache Iceberg 中新增数据。对 Apache Iceberg 这样流批统一的存储层来说，Apache Flink 是真正意义上第一个实现了流批读写  Iceberg 的计算引擎，这也标志着 Apache Flink 和 Apache Iceberg  在共同打造流批统一的数据湖架构上开启了新的篇章。

相关组件版本：

HDFS:3.0.0-CDH6.2.1

Hive:2.1.1-CDH6.2.1

Flink:1.11.1

Iceberg:0.11.0

通过Flink SQL Client 流式读取 Iceberg

过程大体参照 Iceberg-Flink官方文档 

 https://github.com/apache/iceberg/blob/master/site/docs/flink.md

Step 1:解压Flink，基于Hadoop环境启动Standalone的Flink集群

1. tar xzvf flink-1.11.1-bin-scala_2.11.tgz

2. export HADOOP_CLASSPATH=`$HADOOP_HOME/bin/hadoop classpath`

3. ./bin/start-cluster.sh

Step 2：启动Flink SQL Client

如果Iceberg Catalog 为 Hadoop

export HADOOP_CLASSPATH=`$HADOOP_HOME/bin/hadoop classpath`

./bin/sql-client.sh embedded -j <flink-runtime-directory>/iceberg-flink-runtime-0.11.0.jar shell

如果Iceberg Catalog 为 Hive(后面测试基于Iceberg Hive Catalog为例)

export HADOOP_CLASSPATH=`$HADOOP_HOME/bin/hadoop classpath`

./bin/sql-client.sh embedded \

    -j <flink-runtime-directory>/iceberg-flink-runtime-0.11.0.jar \

    -j <hive-bundlded-jar-directory>/flink-sql-connector-hive-2.2.0_2.11-1.11.0.jar \

    -j <hive-bundlded-jar-directory>/flink-connector-hive_2.11-1.11.1.jar \

    -j <hive-bundlded-jar-directory>/hive-exec-2.1.1-cdh6.2.1.jar \

    shell

jar包可以去maven仓库下载，或者通过IDEA用Maven直接下载（CDH要配置CDH的repository）

https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-hive-2.2.0_2.11/

<repository>

    <id>cloudera</id>

    <url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>

</repository>

如果这里按Iceberg官网，不加 flink-connector-hive_2.11-1.11.1.jar 和 hive-exec-2.1.1-cdh6.2.1.jar依赖，查询时会报错。

通过Flink集成Hive的官方文档 

https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/hive/

提示增加上述两个依赖后问题解决。

Step 3:创建Iceberg Hive Catalog

CREATE CATALOG iceberg_catalog WITH (

  'type'='iceberg',

  'catalog-type'='hive',

  'uri'='thrift://node103:9083',

  'clients'='5',

  'property-version'='1',

  'hive-conf-dir'='/etc/hive/conf.cloudera.hive');

Step 4：开启对Iceberg表的实时查询

1. use catalog iceberg_catalog;

2. create database iceberg；

3. use iceberg;

4. SET execution.type = streaming;

5. SET table.dynamic-table-options.enabled=true;

6. SELECT * FROM sample2 /*+ OPTIONS('streaming'='true', 'monitor-interval'='1s')*/ ;

7. 启动Flink任务向Iceberg实时写入数据