热门标签
热门文章
- 1鸿蒙（API 12 Beta2版）媒体开发【Audio Kit简介】音频服务
- 2六十五、Spark-综合案例(搜狗搜索日志分析)_spark搜狗搜索引擎日志下载
- 3mysql的锁_mysql加行级写锁后,其他事务还能读取相应行的数据
- 4Caused by: java.lang.IllegalArgumentException: Unsupported class file major version 61
- 5损失函数总结（二）：L1Loss、MSELoss_l1损失函数表达式
- 6https,udp,socket区别_sslsocket和socket区别
- 7Python Qt GUI设计：QTabWidget、QStackedWidget和QDockWidget容器控件类（提升篇—2）_python qtabwidget
- 8Big Data Tools安装_hadoop.dll下载
- 9关闭iis 错误页面显示详细内容
- 10【雕爷学编程】Arduino智能家居之图像识别并执行相应控制操作_arduino视觉识别
当前位置: article > 正文
第八章 Flink集成Iceberg的DataStreamAPI、TableSQLAPI详解_flink1.14.2版本与iceberg0.12.1版本
作者：爱喝兽奶帝天荒 | 2024-08-12 17:02:54
赞
踩
flink1.14.2版本与iceberg0.12.1版本
1、概述
目前Flink支持使用DataStream API 和SQL API方式实时读取和写入I=ceberg表，建议使用SQL API方式实时读取和写入Iceberg表。
- Iceberg支持的Flink版本为1.11.x版本以上，以下为版本匹配关系：
| Flink版本 | Iceberg版本 | 备注 |
|---|---|---|
| Flink1.11.X | Iceberg0.11.1 | |
| Flink1.12.x ~ Flink1.13.x | Iceberg0.12.1 | SQL API有Bug |
| Flink1.14.x | Iceberg0.12.1 | SQL API有Bug |
本次学习以Flink和Iceberg整合使用Flink版本为1.14.5，Iceberg版本为0.12.1版本。
2、DataStream API
2.1、实时写入Iceberg表
2.1.1、导入依赖
<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>org.example</groupId> <artifactId>flinkiceberg1</artifactId> <version>1.0-SNAPSHOT</version> <properties> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <maven.compiler.source>1.8</maven.compiler.source> <maven.compiler.target>1.8</maven.compiler.target> <!-- flink 1.12.x -1.13.x 版本与Iceberg 0.12.1 版本兼容 ，不能与Flink 1.14 兼容--> <flink.version>1.13.5</flink.version> <!--<flink.version>1.12.1</flink.version>--> <!--<flink.version>1.14.2</flink.version>--> <!-- flink 1.11.x 与Iceberg 0.11.1 合适--> <!--<flink.version>1.11.6</flink.version>--> <hadoop.version>3.1.1</hadoop.version> </properties> <dependencies> <dependency> <groupId>com.alibaba.ververica</groupId> <artifactId>ververica-connector-iceberg</artifactId> <version>1.13-vvr-4.0.7</version> <exclusions> <exclusion> <groupId>com.google.guava</groupId> <artifactId>guava-parent</artifactId> </exclusion> </exclusions> </dependency> <!-- Flink 操作Iceberg 需要的Iceberg依赖 --> <dependency> <groupId>org.apache.iceberg</groupId> <artifactId>iceberg-flink-runtime</artifactId> <version>0.12.1</version> <!--<version>0.11.1</version>--> </dependency> <!-- java开发Flink所需依赖 --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-java</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java_2.11</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients_2.11</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-scala_2.11</artifactId> <version>${flink.version}</version> </dependency> <!-- Flink Kafka连接器的依赖 --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka_2.11</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-base</artifactId> <version>${flink.version}</version> </dependency> <!-- 读取hdfs文件需要jar包--> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>${hadoop.version}</version> <exclusions> <exclusion> <artifactId>guava</artifactId> <groupId>com.google.guava</groupId> </exclusion> </exclusions> </dependency> <!-- Flink SQL & Table--> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-runtime-blink_2.11</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-common</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-api-java</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-api-java-bridge_2.11</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-planner_2.11</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-planner-blink_2.11</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>4.11</version> <scope>test</scope> </dependency> <!-- log4j 和slf4j 包,如果在控制台不想看到日志，可以将下面的包注释掉--> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-log4j12</artifactId> <version>1.7.25</version> <scope>test</scope> </dependency> <dependency> <groupId>log4j</groupId> <artifactId>log4j</artifactId> <version>1.2.17</version> </dependency> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-api</artifactId> <version>1.7.25</version> </dependency> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-nop</artifactId> <version>1.7.25</version> <scope>test</scope> </dependency> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-simple</artifactId> <version>1.7.5</version> </dependency> </dependencies> </project>
- 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
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
- 98
- 99
- 100
- 101
- 102
- 103
- 104
- 105
- 106
- 107
- 108
- 109
- 110
- 111
- 112
- 113
- 114
- 115
- 116
- 117
- 118
- 119
- 120
- 121
- 122
- 123
- 124
- 125
- 126
- 127
- 128
- 129
- 130
- 131
- 132
- 133
- 134
- 135
- 136
- 137
- 138
- 139
- 140
- 141
- 142
- 143
- 144
- 145
- 146
- 147
- 148
- 149
- 150
- 151
- 152
- 153
- 154
- 155
- 156
- 157
- 158
- 159
- 160
- 161
- 162
- 163
- 164
- 165
- 166
- 167
- 168
- 169
- 170
- 171
2.1.2、创建Iceberg表
- 核心：通过Flink创建Iceberg表
-- 1、创建catalog CREATE CATALOG hadoop_catalog WITH ( > 'type'='iceberg', > 'catalog-type'='hadoop', > 'warehouse'='hdfs://leidi01:8020/iceberg_catalog', > 'property-version'='1' > ); -- 2、创建databases create database flink_iceberg; -- 3、创建Sink表 CREATE TABLE hadoop_catalog.flink_iceberg.icebergdemo1 ( id STRING, data STRING );
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 运行结果
2.1.3、代码实现
public class FlinkIcebergDemo1 { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //1.必须设置checkpoint ,Flink向Iceberg中写入数据时当checkpoint发生后，才会commit数据。 env.enableCheckpointing(5000); //2.读取Kafka 中的topic 数据 KafkaSource<String> source = KafkaSource.<String>builder() .setBootstrapServers("192.168.6.102:6667") .setTopics("json") .setGroupId("my-group-id") .setStartingOffsets(OffsetsInitializer.latest()) .setValueOnlyDeserializer(new SimpleStringSchema()) .build(); DataStreamSource<String> kafkaSource = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source"); //3.对数据进行处理，包装成RowData 对象，方便保存到Iceberg表中。 SingleOutputStreamOperator<RowData> dataStream = kafkaSource.map(new MapFunction<String, RowData>() { @Override public RowData map(String s) throws Exception { System.out.println("s = "+s); String[] split = s.split(","); GenericRowData row = new GenericRowData(4); row.setField(0, Integer.valueOf(split[0])); row.setField(1, StringData.fromString(split[1])); row.setField(2, Integer.valueOf(split[2])); row.setField(3, StringData.fromString(split[3])); return row; } }); //4.创建Hadoop配置、Catalog配置和表的Schema，方便后续向路径写数据时可以找到对应的表 Configuration hadoopConf = new Configuration(); Catalog catalog = new HadoopCatalog(hadoopConf,"hdfs://leidi01:8020/flinkiceberg/"); //配置iceberg 库名和表名 TableIdentifier name = TableIdentifier.of("icebergdb", "flink_iceberg_tbl"); //创建Icebeng表Schema Schema schema = new Schema( Types.NestedField.required(1, "id", Types.IntegerType.get()), Types.NestedField.required(2, "nane", Types.StringType.get()), Types.NestedField.required(3, "age", Types.IntegerType.get()), Types.NestedField.required(4, "loc", Types.StringType.get())); //如果有分区指定对应分区，这里“loc”列为分区列，可以指定unpartitioned 方法不设置表分区 // PartitionSpec spec = PartitionSpec.unpartitioned(); PartitionSpec spec = PartitionSpec.builderFor(schema).identity("loc").build(); //指定Iceberg表数据格式化为Parquet存储 Map<String, String> props = ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, FileFormat.PARQUET.name()); Table table = null; // 通过catalog判断表是否存在，不存在就创建，存在就加载 if (!catalog.tableExists(name)) { table = catalog.createTable(name, schema, spec, props); }else { table = catalog.loadTable(name); } TableLoader tableLoader = TableLoader.fromHadoopTable("hdfs://leidi01:8020/flinkiceberg//icebergdb/flink_iceberg_tbl", hadoopConf); //5.通过DataStream Api 向Iceberg中写入数据 FlinkSink.forRowData(dataStream) //这个 .table 也可以不写，指定tableLoader 对应的路径就可以。 .table(table) .tableLoader(tableLoader) //默认为false,追加数据。如果设置为true 就是覆盖数据 .overwrite(false) .build(); env.execute("DataStream Api Write Data To Iceberg"); } }
- 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
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 注意事项：
（1）需要设置Checkpoint，Flink向Iceberg中写入Commit数据时，只有Checkpoint成功之后才会Commit数据，否则后期在Hive中查询不到数据。
（2）读取Kafka数据后需要包装成RowData或者Row对象，才能向Iceberg表中写出数据。写出数据时默认是追加数据，如果指定overwrite就是全部覆盖数据。
（3）在向Iceberg表中写数据之前需要创建对应的Catalog、表Schema,否则写出时只指定对应的路径会报错找不到对应的Iceberg表。
（4）不建议使用DataStream API 向Iceberg中写数据，建议使用SQL API。
2.1.4、Kafka消费者启动
bin/kafka-console-producer.sh --topic json --broker-list leidi01:6667
bin/kafka-console-consumer.sh --bootstrap-server leidi01:6667 --topic json --from-beginning
- 1
- 2
- 3
- 生产数据
- 运行结果：
data中有两个分区
2.1.5、查询表结果
- 说明：
在Flink SQL中创建Hadoop Catalog。
-- 1、创建Hadoop Catalog
CREATE CATALOG flinkiceberg WITH (
'type'='iceberg',
'catalog-type'='hadoop',
'warehouse'='hdfs://leidi01:8020/flinkiceberg/',
'property-version'='1'
);
-- 2、查询表中数据
use catalog flinkiceberg;
use icebergdb;
select * from flink_iceberg_tbl;
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 运行结果
2.2、批量/实时读取Iceberg表
- 核心：DataStream API 读取Iceberg表又分为批量读取和实时读取，通过方法“streaming(true/false)”来控制。
2.2.1、批量读取
- 说明：设置方法“streaming(false)”
- 代码实现
public class FlinkIcebergRead { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //1.配置TableLoader Configuration hadoopConf = new Configuration(); TableLoader tableLoader = TableLoader.fromHadoopTable("hdfs://leidi01:8020/flinkiceberg//icebergdb/flink_iceberg_tbl", hadoopConf); //2.从Iceberg中读取全量/增量读取数据 DataStream<RowData> batchData = FlinkSource.forRowData().env(env) .tableLoader(tableLoader) //默认为false,整批次读取，设置为true 为流式读取 .streaming(false) .build(); batchData.map(new MapFunction<RowData, String>() { @Override public String map(RowData rowData) throws Exception { int id = rowData.getInt(0); String name = rowData.getString(1).toString(); int age = rowData.getInt(2); String loc = rowData.getString(3).toString(); return id+","+name+","+age+","+loc; } }).print(); env.execute("DataStream Api Read Data From Iceberg"); } }
- 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
- 运行结果
2.2.2、实时读取
-
说明：设置方法“streaming(true)”
-
代码实现
DataStream<RowData> batchData = FlinkSource.forRowData().env(env)
.tableLoader(tableLoader)
//默认为false,整批次读取，设置为true 为流式读取
.streaming(true)
.build();
- 1
- 2
- 3
- 4
- 5
- Flink SQL插入数据
insert into flink_iceberg_tbl values (5,'s1',30,'guangzhou'),(6,'s2',31,'tianjin');
- 1
- 运行结果
2.2.3、指定基于快照实时增量读取数据
- 核心：设置方法StartSnapshotId(快照编号)
（1）查看快照编号
（2）代码实现
//2.从Iceberg中读取全量/增量读取数据
DataStream<RowData> batchData = FlinkSource.forRowData().env(env)
.tableLoader(tableLoader)
//基于某个快照实时增量读取数据，快照需要从元数据中获取
.startSnapshotId(1738199999360637062L)
//默认为false,整批次读取; 设置为true为流式读取
.streaming(true)
.build();
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
（3）运行结果
- 说明：*只读取到指定快照往后的数据*
2.2.4、合并Data Flies
- 说明：Iceberg提供Api通过定期提交任务将小文件合并成大文件，可以通过Flink 批任务来执行。
（1）未处理文件
- 说明：Iceberg每提交一次数据都会产生一个Data File。
（2）代码实现
public class RewrietDataFiles { public static void main(String[] args) { ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); // 1、配置TableLoader Configuration hadoopConf = new Configuration(); //2.创建Hadoop配置、Catalog配置和表的Schema，方便后续向路径写数据时可以找到对应的表 Catalog catalog = new HadoopCatalog(hadoopConf,"hdfs://leidi01:8020/flinkiceberg/"); //3.配置iceberg 库名和表名并加载表 TableIdentifier name = TableIdentifier.of("icebergdb", "flink_iceberg_tbl"); Table table = catalog.loadTable(name); //4..合并 data files 小文件 RewriteDataFilesActionResult result = Actions.forTable(table) .rewriteDataFiles() //默认 512M ，可以手动通过以下指定合并文件大小，与Spark中一样。 .targetSizeInBytes(536870912L) .execute(); } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
（3）运行结果
3、SQL API
3.1、创建表并插入数据
（1）代码实现
public class SQLAPIWriteIceberg { public static void main(String[] args) { 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://leidi01:8020/flinkiceberg')"); //2.使用当前Catalog tblEnv.useCatalog("hadoop_iceberg"); //3.创建数据库 tblEnv.executeSql("create database iceberg_db"); //4.使用数据库 tblEnv.useDatabase("iceberg_db"); //5.创建iceberg表 flink_iceberg_tbl tblEnv.executeSql("create table hadoop_iceberg.iceberg_db.flink_iceberg_tbl2(id int,name string,age int,loc string) partitioned by (loc)"); //6.写入数据到表 flink_iceberg_tbl tblEnv.executeSql("insert into hadoop_iceberg.iceberg_db.flink_iceberg_tbl2 values (1,'zs',18,'beijing'),(2,'ls',19,'shanghai'),(3,'ww',20,'guangzhou')"); } }
- 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
（2）运行结果
- 说明：通过HDFS查看文件是否生成。
（3）查看数据
- 说明：通过FlinkSQL查看表中数据
-- 1、创建Catalog
CREATE CATALOG flinkiceberg WITH (
> 'type'='iceberg',
> 'catalog-type'='hadoop',
> 'warehouse'='hdfs://leidi01:8020/flinkiceberg/',
> 'property-version'='1'
> );
-- 2、查询数据
use catalog flinkiceberg
use iceberg_db;
select * from flink_iceberg_tbl2;
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 查看结果
3.2、批量查询表数据
- 说明：SQL API批量查询表中数据，直接查询显示即可
（1）代码逻辑
（2）代码实现
public class SQLAPIReadIceberg { public static void main(String[] args) { 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://leidi01:8020/flinkiceberg')"); //2.批量读取表数据 TableResult tableResult = tblEnv.executeSql("select * from hadoop_iceberg.iceberg_db.flink_iceberg_tbl2 "); tableResult.print(); } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 运行结果
3.3、实时查询表数据
- 说明：link SQL API 实时查询Iceberg表数据时需要设置参数**“table.dynamic-table-options.enabled”为true**，以支持SQL语法中的“OPTIONS”选项
（1）代码逻辑
（2）代码实现
public class SQLStreamReadIceberg { public static void main(String[] args) { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); StreamTableEnvironment tblEnv = StreamTableEnvironment.create(env); env.enableCheckpointing(1000); Configuration configuration = tblEnv.getConfig().getConfiguration(); // 支持SQL语法中的 OPTIONS 选项 configuration.setBoolean("table.dynamic-table-options.enabled", true); //1.创建Catalog tblEnv.executeSql("CREATE CATALOG hadoop_iceberg WITH (" + "'type'='iceberg'," + "'catalog-type'='hadoop'," + "'warehouse'='hdfs://leidi01:8020/flinkiceberg')"); //2.从Iceberg表当前快照读取所有数据，并继续增量读取数据 // streaming指定为true支持实时读取数据，monitor_interval 监控数据的间隔，默认1s TableResult tableResult = tblEnv.executeSql("select * from hadoop_iceberg.iceberg_db.flink_iceberg_tbl2 /*+ 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
- 运行结果：
（3）测试验证
- FlinkSQL插入数据
insert into flink_iceberg_tbl2 values (5,'s1',30,'guangzhou'),(6,'s2',31,'tianjin');
- 1
- 运行结果：在IDEA的控制台可以看到新增数据
3.4、基于快照实时增量读取数据
- 说明：基于某个snapshot-id来继续实时获取数据
（1）代码逻辑
（2）代码实现
- FlinkSQL插入数据
insert into flink_iceberg_tbl2 values (7,'s11',30,'beijing'),(8,'s22',31,'beijing');
- 1
- snapshot-id如下：
- 代码实现
public class SQLSnapshotReadIceberg { public static void main(String[] args) { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); StreamTableEnvironment tblEnv = StreamTableEnvironment.create(env); env.enableCheckpointing(1000); Configuration configuration = tblEnv.getConfig().getConfiguration(); // 支持SQL语法中的 OPTIONS 选项 configuration.setBoolean("table.dynamic-table-options.enabled", true); //1.创建Catalog tblEnv.executeSql("CREATE CATALOG hadoop_iceberg WITH (" + "'type'='iceberg'," + "'catalog-type'='hadoop'," + "'warehouse'='hdfs://leidi01:8020/flinkiceberg')"); //2.从Iceberg 指定的快照继续实时读取数据，快照ID从对应的元数据中获取 //start-snapshot-id :快照ID TableResult tableResult2 = tblEnv.executeSql("SELECT * FROM hadoop_iceberg.iceberg_db.flink_iceberg_tbl2 /*+ OPTIONS('streaming'='true', 'monitor-interval'='1s', 'start-snapshot-id'='8334669420406375204')*/"); tableResult2.print(); } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
（3）运行结果
4、常见报错
4.1、window远程连接hadoop环境变量找不到
- 报错日志
HADOOP_HOME and hadoop.home.dir are unset.
- 1
- 报错原因：本地远程连接Hadoop系统时需要在本地配置相关的Hadoop变量，主要包括hadoop.dll 与 winutils.exe 等。
winutils：由于hadoop主要基于linux编写，**winutil.exe主要用于模拟linux下的目录环境**。当Hadoop在windows下运行或调用远程Hadoop集群的时候，需要该辅助程序才能运行。winutils是Windows中的二进制文件，适用于不同版本的Hadoop系统并构建在Windows VM上，该VM用以在Windows系统中测试Hadoop相关的应用程序。
- 1
- 解决方案：
（1）下载hadoop集群对应winutils版本
- 注意事项：如果你安装的hadoop版本是：3.1.2或者3.2.0 就用winutils-master里面的hadoop-3.0.0配置环境变量吧！
https://github.com/steveloughran/winutils
- 1
（2）将环境变量%HADOOP_HOME%设置为指向包含WINUTILS.EXE的BIN目录上方的目录
4.2、guava包版本冲突
- 报错日志
com.google.common.base.Preconditions.checkArgument(ZLjava/lang/String;Ljava/lang/Object;)V
- 1
- 报错原因：
guava包版本冲突 - 解决方案：使用Maven Helper插件解决冲突
①第一步：在pom界面点击Dependency Analyzer
②第二步：查看Dependency Analyzer功能界面
Ⅰ、显示冲突的jar包
Ⅱ、以列表形式显示所有依赖
Ⅲ、以数的形式显示所有依赖
③第三步：逐个解决conflicts列表中的jar包冲突问题，以guava为例：
点击guava，找到右侧部分红色字体，即依赖冲突的地方，下图显示当前guava版本是24.0，但是有两个依赖的guava版本分别是27.0.0.1和16.0.1。
④将低版本依赖都排除掉
选中红色字体显示的内容->右键->Exclude，完成上述步骤结果如下：
⑤重新加载依赖配置
-------------------------------------------------------------------分割线-------------------------------------------------------------------------------
以上guava包冲突解决后依旧报错，将Hadoop版本从3.2.2降低到3.1.1不报错。
注意hive-3.1.2依赖的Hadoop版本是3.1.0 [3]，一般不建议runtime的Hadoop版本高于hive依赖的版本。
Ⅰ、解决方法一是在hive-exec里对guava做迁移，这个需要自己手动给hive-exec重新打包。
Ⅱ、解决方法二是降低Hadoop版本，这里不一定要降低集群的Hadoop版本，而只是降低flink和hive这边用到的Hadoop版本，相对于用老的Hadoop客户端去访问新的Hadoop服务器，这个小版本的包容性一般来说是没有问题的。
<hadoop.version>3.2.2</hadoop.version>
<!-->将hadoop版本由3.2.2版本降低为3.1.1<-->
<hadoop.version>3.1.1</hadoop.version>
- 1
- 2
- 3
4.4、log4j2配置文件报错
- 报错日志
ERROR StatusLogger No log4j2 configuration file found. Using default configuration: logging only errors to the console. Set system property 'org.apache.logging.log4j.simplelog.StatusLogger.level' to TRACE to show Log4j2 internal initialization logging.
- 1
- 报错原因：没有发现log4j2配置文件
- 解决方案：添加配置log4j2.xml文件，对应org.apache.logging.log4j.Logger
<?xml version="1.0" encoding="UTF-8"?> <Configuration status="WARN"> <Properties> <property name="log_level" value="info" /> <Property name="log_dir" value="log" /> <property name="log_pattern" value="[%d{yyyy-MM-dd HH:mm:ss.SSS}] [%p] - [%t] %logger - %m%n" /> <property name="file_name" value="test" /> <property name="every_file_size" value="100 MB" /> </Properties> <Appenders> <Console name="Console" target="SYSTEM_OUT"> <PatternLayout pattern="${log_pattern}" /> </Console> <RollingFile name="RollingFile" filename="${log_dir}/${file_name}.log" filepattern="${log_dir}/$${date:yyyy-MM}/${file_name}-%d{yyyy-MM-dd}-%i.log"> <ThresholdFilter level="DEBUG" onMatch="ACCEPT" onMismatch="DENY" /> <PatternLayout pattern="${log_pattern}" /> <Policies> <SizeBasedTriggeringPolicy size="${every_file_size}" /> <TimeBasedTriggeringPolicy modulate="true" interval="1" /> </Policies> <DefaultRolloverStrategy max="20" /> </RollingFile> <RollingFile name="RollingFileErr" fileName="${log_dir}/${file_name}-warnerr.log" filePattern="${log_dir}/$${date:yyyy-MM}/${file_name}-%d{yyyy-MM-dd}-warnerr-%i.log"> <ThresholdFilter level="WARN" onMatch="ACCEPT" onMismatch="DENY" /> <PatternLayout pattern="${log_pattern}" /> <Policies> <SizeBasedTriggeringPolicy size="${every_file_size}" /> <TimeBasedTriggeringPolicy modulate="true" interval="1" /> </Policies> </RollingFile> </Appenders> <Loggers> <Root level="${log_level}"> <AppenderRef ref="Console" /> <AppenderRef ref="RollingFile" /> <appender-ref ref="RollingFileErr" /> </Root> </Loggers> </Configuration>
- 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
- 47
- 48
- 49
- 50
- 51
4.5、Flink Hive Catalog报错
- 报错日志
Exception in thread "main" java.lang.NoSuchMethodError: org.apache.calcite.sql.parser.SqlParser.config()Lorg/apache/calcite/sql/parser/SqlParser$Config;
- 1
-
报错原因：依赖报错
-
解决方案：将所有依赖切换到2.12，切换
flink-table-api-java-bridge到flink-table-api-scala-bridge_2.12。
声明：本文内容由网友自发贡献，转载请注明出处：【wpsshop博客】
推荐阅读
相关标签
