Iceberg从入门到精通系列之十一:Flink DataStream读取Iceberg表
- 一、完整代码
- 二、效果如下所示
一、完整代码
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.data.RowData;
import org.apache.iceberg.flink.TableLoader;
import org.apache.iceberg.flink.source.FlinkSource;
public class ReadDemo1 {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//1.获取tableLoader
TableLoader tableLoader = TableLoader.fromHadoopTable("hdfs://hadoop1:8020/warehouse/spark-iceberg/default/a");
//2.通过FlinkSource构建数据流
DataStream<RowData> build = FlinkSource.forRowData()
.env(env)
.tableLoader(tableLoader)
.streaming(false) // false batch方式 true streaming方式
// .streaming(true)
// .asOfTimestamp()
// .startSnapshotId()
.build();
//转换为二元组
build.map(r -> Tuple2.of(r.getInt(0),r.getLong(1)))
.returns(Types.TUPLE(Types.INT,Types.LONG))
.print();
env.execute();
}
}
- streaming(false) :false batch方式
- streaming(true):true streaming方式
二、效果如下所示
11> (2,3)
11> (4,5)