当前位置：网站首页>数据数据湖（十九）：SQL API 读取Kafka数据实时写入Iceberg表

数据数据湖（十九）：SQL API 读取Kafka数据实时写入Iceberg表

2022-07-26 11:11:00 【51CTO】

SQL API 读取Kafka数据实时写入Iceberg表

从Kafka中实时读取数据写入到Iceberg表中，操作步骤如下：

一、首先需要创建对应的Iceberg表

       
       StreamExecutionEnvironment 
       
       env 
       
       = 
       
       StreamExecutionEnvironment.
       
       getExecutionEnvironment();
       
       StreamTableEnvironment 
       
       tblEnv 
       
       = 
       
       StreamTableEnvironment.
       
       create(
       
       env);
       
       env.
       
       enableCheckpointing(
       
       1000);
       
       //1.创建Catalog
       
       tblEnv.
       
       executeSql(
       
       "CREATE CATALOG hadoop_iceberg WITH (" 
       
       +
       
       "'type'='iceberg'," 
       
       +
       
       "'catalog-type'='hadoop'," 
       
       +
       
       "'warehouse'='hdfs://mycluster/flink_iceberg')");
       
       //2.创建iceberg表 flink_iceberg_tbl
       
       tblEnv.
       
       executeSql(
       
       "create table hadoop_iceberg.iceberg_db.flink_iceberg_tbl3(id int,name string,age int,loc string) partitioned by (loc)");
      
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.

数据数据湖（十九）：SQL API 读取Kafka数据实时写入Iceberg表_kafka

二、编写代码读取Kafka数据实时写入Iceberg

       
       public 
       
       class 
       
       ReadKafkaToIceberg {
       
       public 
       
       static 
       
       void 
       
       main(
       
       String[] 
       
       args) 
       
       throws 
       
       Exception {
       
       StreamExecutionEnvironment 
       
       env 
       
       = 
       
       StreamExecutionEnvironment.
       
       getExecutionEnvironment();
       
       StreamTableEnvironment 
       
       tblEnv 
       
       = 
       
       StreamTableEnvironment.
       
       create(
       
       env);
       
       env.
       
       enableCheckpointing(
       
       1000);
       
       /**
       
       * 1.需要预先创建 Catalog 及Iceberg表
       
       */
       
       //1.创建Catalog
       
       tblEnv.
       
       executeSql(
       
       "CREATE CATALOG hadoop_iceberg WITH (" 
       
       +
       
       "'type'='iceberg'," 
       
       +
       
       "'catalog-type'='hadoop'," 
       
       +
       
       "'warehouse'='hdfs://mycluster/flink_iceberg')");
       
       //2.创建iceberg表 flink_iceberg_tbl
       
       // tblEnv.executeSql("create table hadoop_iceberg.iceberg_db.flink_iceberg_tbl3(id int,name string,age int,loc string) partitioned by (loc)");
       
       //3.创建 Kafka Connector,连接消费Kafka中数据
       
       tblEnv.
       
       executeSql(
       
       "create table kafka_input_table(" 
       
       +
       
       " id int," 
       
       +
       
       " name varchar," 
       
       +
       
       " age int," 
       
       +
       
       " loc varchar" 
       
       +
       
       ") with (" 
       
       +
       
       " 'connector' = 'kafka'," 
       
       +
       
       " 'topic' = 'flink-iceberg-topic'," 
       
       +
       
       " 'properties.bootstrap.servers'='node1:9092,node2:9092,node3:9092'," 
       
       +
       
       " 'scan.startup.mode'='latest-offset'," 
       
       +
       
       " 'properties.group.id' = 'my-group-id'," 
       
       +
       
       " 'format' = 'csv'" 
       
       +
       
       ")");
       
       //4.配置 table.dynamic-table-options.enabled
       
       Configuration 
       
       configuration 
       
       = 
       
       tblEnv.
       
       getConfig().
       
       getConfiguration();
       
       // 支持SQL语法中的 OPTIONS 选项
       
       configuration.
       
       setBoolean(
       
       "table.dynamic-table-options.enabled", 
       
       true);
       
       //5.写入数据到表 flink_iceberg_tbl3
       
       tblEnv.
       
       executeSql(
       
       "insert into hadoop_iceberg.iceberg_db.flink_iceberg_tbl3 select id,name,age,loc from kafka_input_table");
       
       //6.查询表数据
       
       TableResult 
       
       tableResult 
       
       = 
       
       tblEnv.
       
       executeSql(
       
       "select * from hadoop_iceberg.iceberg_db.flink_iceberg_tbl3 /*+ OPTIONS('streaming'='true', 'monitor-interval'='1s')*/");
       
       tableResult.
       
       print();
       
    }
       
}
      
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.

数据数据湖（十九）：SQL API 读取Kafka数据实时写入Iceberg表_kafka_02

启动以上代码，向Kafka topic中生产如下数据：

       
       1,zs,18,beijing
       
2,ls,19,shanghai
       
3,ww,20,beijing
       
4,ml,21,shanghai
      
1.
2.
3.
4.

数据数据湖（十九）：SQL API 读取Kafka数据实时写入Iceberg表_flink_03

我们可以看到控制台上有对应实时数据输出，查看对应的Icberg HDFS目录，数据写入成功。

原网站

版权声明
本文为[51CTO]所创，转载请带上原文链接，感谢
https://blog.51cto.com/lansonli/5513512

当前位置：网站首页>数据数据湖（十九）：SQL API 读取Kafka数据实时写入Iceberg表

数据数据湖（十九）：SQL API 读取Kafka数据实时写入Iceberg表

SQL API 读取Kafka数据实时写入Iceberg表

一、首先需要创建对应的Iceberg表

二、编写代码读取Kafka数据实时写入Iceberg

边栏推荐

猜你喜欢

随机推荐