Iceberg从入门到精通系列之十一：Flink DataStream读取Iceberg表-Toy模板网

这篇具有很好参考价值的文章主要介绍了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();

    }
}