1.問題描述

在應用flink實時消費kafka數據多端中，一般會使用flink原生的addsink或flinkSQL利用SqlDialect，比如消費kafka數據實時寫入hive和kafka一般用兩種方式：
第一種方式是寫入hive利用SqlDialect，寫入kafka利用flink的旁路輸出流+原生addSink
第二種方式是寫入hive和kafka都利用SqlDialect的方式，將kafka也當作一個劉表

2.第一種方式核心代碼及現狀

	DataStream<String> dataStream = environment.addSource(new FlinkKafkaConsumer(topic, new SimpleStringSchema(), props));
        SingleOutputStreamOperator<SipDataInfo> mainStream = dataStream.map(s -> {
            SipDataInfo sipDataInfo = new SipDataInfo();
            JSONObject jsonObject = SipFullauditMonitor.complex(s);

            sipDataInfo.setRow(createRow(jsonObject, size, typeArray, column));
            sipDataInfo.setJsonObject(jsonObject);
            return sipDataInfo;
        });

        final OutputTag<SipDataInfo> kafkaOutputTag = new OutputTag<SipDataInfo>("kafka_stream") {
        };
        final OutputTag<SipDataInfo> hiveOutputTag = new OutputTag<SipDataInfo>("hive_stream") {
        };

        SingleOutputStreamOperator<SipDataInfo> sideOutputStream = mainStream.process(new ProcessFunction<SipDataInfo, SipDataInfo>() {
            @Override
            public void processElement(SipDataInfo sipDataInfo, Context context, Collector<SipDataInfo> collector) throws Exception {
                context.output(kafkaOutputTag, sipDataInfo);
                context.output(hiveOutputTag, sipDataInfo);
            }
        });

        DataStream<SipDataInfo> kafkaStream = sideOutputStream.getSideOutput(kafkaOutputTag);
        DataStream<SipDataInfo> hiveStream = sideOutputStream.getSideOutput(hiveOutputTag);

        Properties producerProperties = new Properties();
        producerProperties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "ambari1:6667");

        kafkaStream.map(sipDataInfo -> sipDataInfo.getJsonObject().toJSONString())
                .filter(s -> JSONObject.parseObject(s, SipFullauditMonitor.class).getReftaskid() != null && JSONObject.parseObject(s, SipFullauditMonitor.class).getReftaskid() == 0)
                .addSink(new FlinkKafkaProducer<String>("dwd_" + topic, new SimpleStringSchema(), props));


        TypeInformation[] tfs = getSqlColumsType(typeArray);
        DataStream<Row> hiveOdsSinkDataStream = hiveStream.map(sipDataInfo -> sipDataInfo.getRow())
                .returns(Types.ROW_NAMED(column, tfs))
                .filter(row -> CommonUtil.filter(row));

        setHiveParam(parameter, tableEnv);

        Table table = tableEnv.fromDataStream(hiveOdsSinkDataStream);
        tableEnv.createTemporaryView("tmp_" + topic, table);

        tableEnv.getConfig().setSqlDialect(SqlDialect.HIVE);
        tableEnv.executeSql(BaseStreamLaucher.parseCreateTableSqlByColumn("ods_" + topic,column, typeArray,new String[]{"pdate","insterhour"},new String[]{"string","string"}));

        //寫hive表
        tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);
        String insertSql = "insert into ods_" + topic + " PARTITION(pdate='" +
                new SimpleDateFormat("yyyy-MM-dd").format(new Date()) +
                "') select " + sinkHiveColumnStr + " from tmp_" + topic;

        tableEnv.executeSql(insertSql);
        environment.execute();
    }
}

3.第二種方式實現的核心代碼

		DataStream<String> dataStream = environment.addSource(new FlinkKafkaConsumer(topic, new SimpleStringSchema(), props));

        TypeInformation[] tfs = getSqlColumsType(typeArray);
        DataStream<Row> rowDataStream = dataStream.map(s -> createRow(SipFullauditMonitor.complex(s), size, typeArray, column))
                .returns(Types.ROW_NAMED(column, tfs))
                .filter(row -> CommonUtil.filter(row));

        Table table = tableEnv.fromDataStream(rowDataStream);
        setHiveParam(parameter, tableEnv);
        tableEnv.createTemporaryView("tmp_" + topic, table);

        //創建hive表
        tableEnv.executeSql(BaseStreamLaucher.parseCreateTableSqlByColumn("ods_" + topic,column, typeArray,new String[]{"pdate","insterhour"},new String[]{"string","string"}));

        tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);
        //創建kafka表
        tableEnv.executeSql("drop table dwd_sip_fullaudit_monitor");
        String kafkaTableSql = createKafkaTableSqlByColumn("dwd_sip_fullaudit_monitor", parameter, column, typeArray);
        tableEnv.executeSql(kafkaTableSql);

        //寫hive表
        String insertHiveSql = "insert into ods_" + topic + " PARTITION(pdate='" +
                new SimpleDateFormat("yyyy-MM-dd").format(new Date()) +
                "',insterhour='" + new SimpleDateFormat("yyyyMMddHH").format(new Date()) + "') select " + sinkHiveColumnStr + " from tmp_" + topic;
      
        //寫kafka表
        String insertKafkaSql = "insert into dwd_sip_fullaudit_monitor" + " select " + sinkHiveColumnStr + " from " + "tmp_" + topic;

        tableEnv.executeSql(insertKafkaSql);
        tableEnv.executeSql(insertHiveSql);
        

在以上兩種實現方式中，發現flink都會在yarn上啟動兩個應用，這兩個應用雖然都能將數據正常寫入hive和kafka，但是不太好。

后面通過不斷的嘗試api發現StreamTableEnvironment StreamStatementSet可以解決該問題

4.應用StreamTableEnvironment StreamStatementSet的核心代碼

		DataStream<String> dataStream = environment.addSource(new FlinkKafkaConsumer(topic, new SimpleStringSchema(), props));

        TypeInformation[] tfs = getSqlColumsType(typeArray);
        DataStream<Row> rowDataStream = dataStream.map(s -> createRow(SipFullauditMonitor.complex(s), size, typeArray, column))
                .returns(Types.ROW_NAMED(column, tfs))
                .filter(row -> CommonUtil.filter(row));

        Table table = tableEnv.fromDataStream(rowDataStream);
        setHiveParam(parameter, tableEnv);
        tableEnv.createTemporaryView("tmp_" + topic, table);

        //創建hive表
        tableEnv.executeSql(BaseStreamLaucher.parseCreateTableSqlByColumn("ods_" + topic,column, typeArray,new String[]{"pdate","insterhour"},new String[]{"string","string"}));

        tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);
        //創建kafka表
        tableEnv.executeSql("drop table dwd_sip_fullaudit_monitor");
        String kafkaTableSql = createKafkaTableSqlByColumn("dwd_sip_fullaudit_monitor", parameter, column, typeArray);
        tableEnv.executeSql(kafkaTableSql);

        StatementSet stmtSet = tableEnv.createStatementSet();
        //寫hive表
        String insertHiveSql = "insert into ods_" + topic + " PARTITION(pdate='" +
                new SimpleDateFormat("yyyy-MM-dd").format(new Date()) +
                "',insterhour='" + new SimpleDateFormat("yyyyMMddHH").format(new Date()) + "') select " + sinkHiveColumnStr + " from tmp_" + topic;
        System.out.println("insertHiveSql:"+insertHiveSql);
        //寫kafka表
        String insertKafkaSql = "insert into dwd_sip_fullaudit_monitor" + " select " + sinkHiveColumnStr + " from " + "tmp_" + topic;

        stmtSet.addInsertSql(insertHiveSql);
        stmtSet.addInsertSql(insertKafkaSql);

        stmtSet.execute();

執行查看flink web界面

說明：

StreamStatementSet的這個的應用在初學或者一般場景應用下可能不太容易發現或應用，來看下flink源碼的解釋，紅色部分大概意思是[可以一起優化所有添加的語句，然后將它們作為一個作業提交],重點是作為一個作業提交。但StreamStatementSet并沒有解決前面的第一種場景。所以在實際的應用中不太建議流表和原生addsink混用，flink越往后的版本也是更加提倡應用流表方式去完成流批一體的體系