Flink中TableAPI 、 SQL 與 Kafka消息插入是怎么實現(xiàn)的

這期內(nèi)容當(dāng)中小編將會給大家?guī)碛嘘P(guān)Flink中TableAPI 、 SQL 與 Kafka消息插入是怎么實現(xiàn)的，文章內(nèi)容豐富且以專業(yè)的角度為大家分析和敘述，閱讀完這篇文章希望大家可以有所收獲。

使用Tbale&SQL與Flink Kafka連接器將數(shù)據(jù)寫入kafka的消息隊列

示例環(huán)境

java.version: 1.8.xflink.version: 1.11.1kafka：2.11

示例數(shù)據(jù)源 （項目碼云下載）

Flink 系例 之 搭建開發(fā)環(huán)境與數(shù)據(jù)

示例模塊 （pom.xml）

Flink 系例 之 TableAPI & SQL 與 示例模塊

InsertToKafka.java

package com.flink.examples.kafka;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.StatementSet;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Description 使用Tbale&SQL與Flink Elasticsearch連接器將數(shù)據(jù)寫入kafka的消息隊列
 */
public class InsertToKafka {

    /**
        官方參考：https://ci.apache.org/projects/flink/flink-docs-release-1.12/zh/dev/table/connectors/kafka.html

        format:用于反序列化和序列化Kafka消息的格式。支持的格式包括'csv'，'json'，'avro'，'debezium-json'和'canal-json'。
     */

    static String table_sql = "CREATE TABLE KafkaTable (\n" +
            "  `user_id` BIGINT,\n" +
            "  `item_id` BIGINT,\n" +
            "  `behavior` STRING,\n" +
            "  `ts` TIMESTAMP(3)\n" +
            ") WITH (\n" +
            "  'connector' = 'kafka',\n" +
            "  'topic' = 'user_behavior',\n" +
            "  'properties.bootstrap.servers' = '192.168.110.35:9092',\n" +
            "  'properties.group.id' = 'testGroup',\n" +
            "  'scan.startup.mode' = 'earliest-offset',\n" +
            "  'format' = 'json'\n" +
            ")";
    public static void main(String[] args) throws Exception {
        //構(gòu)建StreamExecutionEnvironment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //默認(rèn)流時間方式
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
        //構(gòu)建EnvironmentSettings 并指定Blink Planner
        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        //構(gòu)建StreamTableEnvironment
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, bsSettings);
        //注冊kafka數(shù)據(jù)維表
        tEnv.executeSql(table_sql);

        //時間格式處理，參考阿里文檔
        //https://www.alibabacloud.com/help/zh/faq-detail/64813.htm?spm=a2c63.q38357.a3.3.697c13523NZiIN
        String sql = "insert into KafkaTable (user_id,item_id,behavior,ts) values(1,1,'normal', TO_TIMESTAMP(FROM_UNIXTIME( " + System.currentTimeMillis() + " / 1000, 'yyyy-MM-dd HH:mm:ss')))";

        // 第一種方式：直接執(zhí)行sql
//        TableResult tableResult = tEnv.executeSql(sql);

        //第二種方式：聲明一個操作集合來執(zhí)行sql
        StatementSet stmtSet = tEnv.createStatementSet();
        stmtSet.addInsertSql(sql);
        TableResult tableResult = stmtSet.execute();

        tableResult.print();
    }
}

打印結(jié)果

+-------------------------------------------+
| default_catalog.default_database.my_users |
+-------------------------------------------+
|                                        -1 |
+-------------------------------------------+
1 row in set

上述就是小編為大家分享的Flink中TableAPI 、 SQL 與 Kafka消息插入是怎么實現(xiàn)的了，如果剛好有類似的疑惑，不妨參照上述分析進行理解。如果想知道更多相關(guān)知識，歡迎關(guān)注億速云行業(yè)資訊頻道。