0
0

数据湖（十八）：Flink与Iceberg整合SQL API操作

Flink与Iceberg整合SQL API操作

Flink SQL 在操作Iceberg时，对应的版本为Flink 1.11.x 与Iceberg0.11.1版本，目前，Flink1.14.2版本与Iceberg0.12.1版本对于SQL API 来说兼容有问题，所以这里使用Flink1.11.6版本与Iceberg0.11.1版本来演示Flink SQL API 操作Iceberg。

一、​​​​​​​​​​​​​​SQL API 创建Iceberg表并写入数据

1、创建新项目，导入如下maven依赖包

  1.  <properties>
  2.   <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  3.   <maven.compiler.source>1.8</maven.compiler.source>
  4.   <maven.compiler.target>1.8</maven.compiler.target>
  5.   <!-- flink 1.11.x 与Iceberg 0.11.1 合适-->
  6.   <flink.version>1.11.6</flink.version>
  7.   <hadoop.version>3.2.2</hadoop.version>
  8. </properties>
  10. <dependencies>
  11.   <!-- Flink 操作Iceberg 需要的Iceberg依赖 -->
  12.   <dependency>
  13.     <groupId>org.apache.iceberg</groupId>
  14.     <artifactId>iceberg-flink-runtime</artifactId>
  15.     <version>0.11.1</version>
  16.   </dependency>
  18.   <!-- java 开发Flink 所需依赖 -->
  19.   <dependency>
  20.     <groupId>org.apache.flink</groupId>
  21.     <artifactId>flink-java</artifactId>
  22.     <version>${flink.version}</version>
  23.   </dependency>
  24.   <dependency>
  25.     <groupId>org.apache.flink</groupId>
  26.     <artifactId>flink-streaming-java_2.11</artifactId>
  27.     <version>${flink.version}</version>
  28.   </dependency>
  30.   <dependency>
  31.     <groupId>org.apache.flink</groupId>
  32.     <artifactId>flink-clients_2.11</artifactId>
  33.     <version>${flink.version}</version>
  34.   </dependency>
  36.   <dependency>
  37.     <groupId>org.apache.flink</groupId>
  38.     <artifactId>flink-streaming-scala_2.11</artifactId>
  39.     <version>${flink.version}</version>
  40.   </dependency>
  42.   <!-- Flink Kafka连接器的依赖 -->
  43.   <dependency>
  44.     <groupId>org.apache.flink</groupId>
  45.     <artifactId>flink-connector-kafka_2.11</artifactId>
  46.     <version>${flink.version}</version>
  47.   </dependency>
  48.   <dependency>
  49.     <groupId>org.apache.flink</groupId>
  50.     <artifactId>flink-csv</artifactId>
  51.     <version>${flink.version}</version>
  52.   </dependency>
  54.   <!-- 读取hdfs文件需要jar包-->
  55.   <dependency>
  56.     <groupId>org.apache.hadoop</groupId>
  57.     <artifactId>hadoop-client</artifactId>
  58.     <version>${hadoop.version}</version>
  59.   </dependency>
  61.   <!-- Flink SQL & Table-->
  62.   <dependency>
  63.     <groupId>org.apache.flink</groupId>
  64.     <artifactId>flink-table-runtime-blink_2.11</artifactId>
  65.     <version>${flink.version}</version>
  66.   </dependency>
  68.   <dependency>
  69.     <groupId>org.apache.flink</groupId>
  70.     <artifactId>flink-table</artifactId>
  71.     <version>${flink.version}</version>
  72.   </dependency>
  74.   <dependency>
  75.     <groupId>org.apache.flink</groupId>
  76.     <artifactId>flink-table-common</artifactId>
  77.     <version>${flink.version}</version>
  78.   </dependency>
  80.   <dependency>
  81.     <groupId>org.apache.flink</groupId>
  82.     <artifactId>flink-table-api-java</artifactId>
  83.     <version>${flink.version}</version>
  84.   </dependency>
  85.   <dependency>
  86.     <groupId>org.apache.flink</groupId>
  87.     <artifactId>flink-table-api-java-bridge_2.11</artifactId>
  88.     <version>${flink.version}</version>
  89.   </dependency>
  91.   <dependency>
  92.     <groupId>org.apache.flink</groupId>
  93.     <artifactId>flink-table-planner_2.11</artifactId>
  94.     <version>${flink.version}</version>
  95.   </dependency>
  97.   <dependency>
  98.     <groupId>org.apache.flink</groupId>
  99.     <artifactId>flink-table-planner-blink_2.11</artifactId>
  100.     <version>${flink.version}</version>
  101.   </dependency>
  103.   <dependency>
  104.     <groupId>junit</groupId>
  105.     <artifactId>junit</artifactId>
  106.     <version>4.11</version>
  107.     <scope>test</scope>
  108.   </dependency>
  110.   <!-- log4j 和slf4j 包,如果在控制台不想看到日志，可以将下面的包注释掉-->
  111.   <dependency>
  112.     <groupId>org.slf4j</groupId>
  113.     <artifactId>slf4j-log4j12</artifactId>
  114.     <version>1.7.25</version>
  115.     <scope>test</scope>
  116.   </dependency>
  117.   <dependency>
  118.     <groupId>log4j</groupId>
  119.     <artifactId>log4j</artifactId>
  120.     <version>1.2.17</version>
  121.   </dependency>
  122.   <dependency>
  123.     <groupId>org.slf4j</groupId>
  124.     <artifactId>slf4j-api</artifactId>
  125.     <version>1.7.25</version>
  126.   </dependency>
  127.   <dependency>
  128.     <groupId>org.slf4j</groupId>
  129.     <artifactId>slf4j-nop</artifactId>
  130.     <version>1.7.25</version>
  131.     <scope>test</scope>
  132.   </dependency>
  133.   <dependency>
  134.     <groupId>org.slf4j</groupId>
  135.     <artifactId>slf4j-simple</artifactId>
  136.     <version>1.7.5</version>
  137.   </dependency>
  138. </dependencies>

2、编写Flink SQL 创建Iceberg表并写入数据

  1. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  2.         StreamTableEnvironment tblEnv = StreamTableEnvironment.create(env);
  4.         env.enableCheckpointing(1000);
  6.         //1.创建Catalog
  7.         tblEnv.executeSql("CREATE CATALOG hadoop_iceberg WITH (" +
  8.                 "'type'='iceberg'," +
  9.                 "'catalog-type'='hadoop'," +
  10.                 "'warehouse'='hdfs://mycluster/flink_iceberg')");
  12.         //2.使用当前Catalog
  13.         tblEnv.useCatalog("hadoop_iceberg");
  15.         //3.创建数据库
  16.         tblEnv.executeSql("create database iceberg_db");
  18.         //4.使用数据库
  19.         tblEnv.useDatabase("iceberg_db");
  21.         //5.创建iceberg表 flink_iceberg_tbl
  22.         tblEnv.executeSql("create table hadoop_iceberg.iceberg_db.flink_iceberg_tbl2(id int,name string,age int,loc string) partitioned by (loc)");
  24.         //6.写入数据到表 flink_iceberg_tbl
  25.         tblEnv.executeSql("insert into hadoop_iceberg.iceberg_db.flink_iceberg_tbl2 values (1,'zs',18,'beijing'),(2,'ls',19,'shanghai'),(3,'ww',20,'guangzhou')");

3、在Hive中映射Iceberg表并查询

在Hive中执行如下命令创建对应的Iceberg表：

  1. #在Hive中创建Iceberg表
  2. CREATE TABLE flink_iceberg_tbl2  (
  3.   id int, 
  4.   name string,
  5.   age int,
  6.   loc string
  7. )
  8. STORED BY 'org.apache.iceberg.mr.hive.HiveIcebergStorageHandler' 
  9. LOCATION 'hdfs://mycluster/flink_iceberg/iceberg_db/flink_iceberg_tbl2' 
  10. TBLPROPERTIES ('iceberg.catalog'='location_based_table');
  1. #在Hive中查询Iceberg表中的数据
  2. hive> select * from flink_iceberg_tbl2;
  3. OK
  4. 3    ww    20    guangzhou
  5. 1    zs    18    beijing
  6. 2    ls    19    shanghai

二、​​​​​​​​​​​​​​SQL API 批量查询Iceberg表数据

Flink SQL API 批量查询Iceberg表数据，直接查询显示即可。代码如下：

  1. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  2. StreamTableEnvironment tblEnv = StreamTableEnvironment.create(env);
  4. env.enableCheckpointing(1000);
  6. //1.创建Catalog
  7. tblEnv.executeSql("CREATE CATALOG hadoop_iceberg WITH (" +
  8.     "'type'='iceberg'," +
  9.     "'catalog-type'='hadoop'," +
  10.     "'warehouse'='hdfs://mycluster/flink_iceberg')");
  11. //2.批量读取表数据
  12. TableResult tableResult = tblEnv.executeSql("select * from hadoop_iceberg.iceberg_db.flink_iceberg_tbl2 ");
  14. tableResult.print();

结果如下：

三、​​​​​​​​​​​​​​SQL API 实时查询Iceberg表数据

Flink SQL API 实时查询Iceberg表数据时需要设置参数“table.dynamic-table-options.enabled”为true,以支持SQL语法中的“OPTIONS”选项，代码如下：

  1. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  2. StreamTableEnvironment tblEnv = StreamTableEnvironment.create(env);
  4. env.enableCheckpointing(1000);
  6. Configuration configuration = tblEnv.getConfig().getConfiguration();
  7. // 支持SQL语法中的 OPTIONS 选项
  8. configuration.setBoolean("table.dynamic-table-options.enabled", true);
  10. //1.创建Catalog
  11. tblEnv.executeSql("CREATE CATALOG hadoop_iceberg WITH (" +
  12.         "'type'='iceberg'," +
  13.         "'catalog-type'='hadoop'," +
  14.         "'warehouse'='hdfs://mycluster/flink_iceberg')");
  16. //2.从Iceberg表当前快照读取所有数据，并继续增量读取数据
  17. // streaming指定为true支持实时读取数据，monitor_interval 监控数据的间隔，默认1s
  18. TableResult tableResult = tblEnv.executeSql("select * from hadoop_iceberg.iceberg_db.flink_iceberg_tbl2 /*+ OPTIONS('streaming'='true', 'monitor-interval'='1s')*/");
  20. tableResult.print();

启动以上代码后，可以看到会将目前存在于Iceberg表中的数据读取出来，向Hive中对应的Iceberg表中插入数据，可以看到控制台实时获取数据。

  1. #在向Hive的Iceberg表中插入数据之前需要加入以下两个包：
  2. add jar /software/hive-3.1.2/lib/iceberg-hive-runtime-0.12.1.jar;
  3. add jar /software/hive-3.1.2/lib/libfb303-0.9.3.jar;
  5. #向Hive 中Iceberg 表插入两条数据
  6. hive> insert into flink_iceberg_tbl2 values (4,'ml',30,'shenzhen'),(5,'tq',31,'beijing');

在控制台可以看到实时新增数据

四、​​​​​​​​​​​​​​SQL API指定基于快照实时增量读取数据

Flink SQL API 还支持基于某个snapshot-id来继续实时获取数据，代码如下：

  1. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  2. StreamTableEnvironment tblEnv = StreamTableEnvironment.create(env);
  3. env.enableCheckpointing(1000);
  5. Configuration configuration = tblEnv.getConfig().getConfiguration();
  6. // 支持SQL语法中的 OPTIONS 选项
  7. configuration.setBoolean("table.dynamic-table-options.enabled", true);
  9. //1.创建Catalog
  10. tblEnv.executeSql("CREATE CATALOG hadoop_iceberg WITH (" +
  11.       "'type'='iceberg'," +
  12.       "'catalog-type'='hadoop'," +
  13.       "'warehouse'='hdfs://mycluster/flink_iceberg')");
  15. //2.从Iceberg 指定的快照继续实时读取数据，快照ID从对应的元数据中获取
  16. //start-snapshot-id :快照ID
  17. TableResult tableResult2 = tblEnv.executeSql("SELECT * FROM hadoop_iceberg.iceberg_db.flink_iceberg_tbl2 /*+ OPTIONS('streaming'='true', 'monitor-interval'='1s', 'start-snapshot-id'='3821550127947089987')*/");
  18. tableResult2.print();
  • 📢博客主页：https://lansonli.blog.csdn.net
  • 📢欢迎点赞 👍 收藏 ⭐留言 📝 如有错误敬请指正！
  • 📢本文由 Lansonli 原创，首发于 CSDN博客🙉
  • 📢停下休息的时候不要忘了别人还在奔跑，希望大家抓紧时间学习，全力奔赴更美好的生活✨
标签： flink sql 大数据
本文转载自: https://blog.csdn.net/xiaoweite1/article/details/125774962
版权归原作者 Lansonli 所有， 如有侵权，请联系我们删除。
登录后发布评论

“数据湖（十八）：Flink与Iceberg整合SQL API操作”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？