[大数据]Flink个人学习整理-Join和Hive Catalog篇（十五）

Flink个人学习整理-Join和Hive Catalog篇（十五）

Flink Join

FlinkSQL中的状态永久保存 默认：PT0S 永久保存

public class FlinkSQL06_Join {
    public static void main(String[] args) throws Exception {
        //  获取运行时环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        //  默认FlinkSQL中的状态永久保存 PT0S
        tableEnvironment.getConfig().getIdleStateRetention();

        //  设置 FlinkSQL中的状态保存10S --（数据空闲10S）
       tableEnvironment.getConfig().setIdleStateRetention(Duration.ofSeconds(10));

        //  TableA
        SingleOutputStreamOperator<TableA> aDS = env.socketTextStream("localhost", 9999)
                .map(line -> {
                    String[] strings = line.split(",");
                    return new TableA(strings[0], strings[1]);
                });

        //  TableB
        SingleOutputStreamOperator<TableB> bDS = env.socketTextStream("localhost", 9998)
                .map(line -> {
                    String[] strings = line.split(",");
                    return new TableB(strings[0],Integer.parseInt(strings[1]));
                });

        //  将流数据转换为表，并创建表名
        tableEnvironment.createTemporaryView("tableA",aDS);
        tableEnvironment.createTemporaryView("tableB",bDS);

        //  双流Join
        tableEnvironment.sqlQuery(" select * from tableA a join tableB b on a.id = b.id")
                .execute()
                .print();

        env.execute();

    }
}

Hive Catalog

public class FlinkSQL01_Catalog {
    public static void main(String[] args) throws Exception {

        //  获取运行时环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        //  创建HiveCatalog
        //  名称 - 库名 - 配置信息路径
        HiveCatalog hiveCatalog = new HiveCatalog("myclusterHive", "default", "input");

        //  注册Catalog
        tableEnvironment.registerCatalog("myclusterHive",hiveCatalog);

        //  使用Catalog
        tableEnvironment.useCatalog("myclusterHive");

        //  执行SQL，操作Hive数据
        tableEnvironment.executeSql("show tables")
                .print();

    }
}