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(); }}
评论