Help Center/ MapReduce Service/ Developer Guide (LTS)/ Flink Development Guide (Normal Mode)/ Developing an Application/ Flink Reads and Writes Iceberg/ Java Sample Code
Updated on 2026-06-29 GMT+08:00
View PDF
Share

Java Sample Code

Description

Call Flink APIs to read data from and write data to Iceberg.

Sample Code

The following example shows the main logic code of ReadFromIcebergToIceberg.

Complete code is provided in com.huawei.bigdata.flink.examples.ReadFromIcebergToIceberg.

Main logic code of ReadFromIcebergToIceberg 
public class ReadFromIcebergToIceberg {
    public static void main(String[] args) throws Exception {
        System.out.println("use command as: ");
        System.out.println(
                "flink run --class com.huawei.bigdata.flink.examples.ReadFromIcebergToIceberg /opt/FlinkIcebergExample.jar "
                        + "  --hive.metastore.uris thrift://xxx.xxx.xxx.xxx:xx,thrift://xxx.xxx.xxx.xxx:xx "        // Hive Metastore address, which is the value of hive.metastore.uris in the hive-site.xml configuration file.
                        + "  --sourceTableName xxx  "    //<sourceTableName> is the name of the Iceberg source table to be read. The default table name is icebergsource.
                        + "  --sinkTableName xxx  "    //<sinkTableName> is the name of the Iceberg target table to be written. The default table name is icebergsink.
                        + "  --hive.metastore.kerberos.principal xxxx  ");   // Hive Metastore principle, which is the value of hive.metastore.kerberos.principal in the hive-site.xml configuration file.

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000);

        ParameterTool paraTool = ParameterTool.fromArgs(args);
        String catalogName = paraTool.get("catalogName", "hive_iceberg_catalog");
        String hiveMetastoreUri = paraTool.get("hive.metastore.uris"); 
        String sourceName = paraTool.get("sourceTableName", "icebergsource");
        String sinkName = paraTool.get("sinkTableName", "icebergsink");
        String hiveMetastorePrincipal = paraTool.get("hive.metastore.kerberos.principal"); 
        String warehouse = paraTool.get("warehouse", "hdfs://hacluster/user/hive/warehouse"); 

        Map<String, String> catalogProps = new HashMap<>();
        catalogProps.put("type", "hive");
        catalogProps.put("uri", hiveMetastoreUri);
        catalogProps.put("warehouse", warehouse);
        Configuration conf = new Configuration();
        conf.set(HiveConf.ConfVars.METASTORE_USE_THRIFT_SASL.varname, "true");
        conf.set(HiveConf.ConfVars.METASTORE_KERBEROS_PRINCIPAL.varname, hiveMetastorePrincipal);
        CatalogLoader catalogLoader = CatalogLoader.hive(catalogName, conf, catalogProps);
        TableIdentifier table = TableIdentifier.of("default", sourceName);
        TableLoader tableLoader = TableLoader.fromCatalog(catalogLoader, table);

        DataStream<RowData> batch = FlinkSource.forRowData()
                .env(env)
                .tableLoader(tableLoader)
                .streaming(false)
                .build();

        DataStream<RowData> readStreamfile = batch.filter(new FilterFunction<RowData>() {
            @Override
            public boolean filter(RowData rowData) throws Exception {
                return rowData.getInt(2) > 100;
            }
        });

        TableIdentifier table2 = TableIdentifier.of("default", sinkName);
        TableLoader tableLoader2 = TableLoader.fromCatalog(catalogLoader, table2);

        FlinkSink.forRowData(readStreamfile)
                .tableLoader(tableLoader2)
                .overwrite(true)
                .append();

        env.execute();
    }
}