Tableapi & SQL and Kafka message insertion in Flink

Use Tbale&SQL And Flink Kafka The connector writes data to kafka Message queue of

Sample environment

java.version: 1.8.x
flink.version: 1.11.1
kafka：2.11

Sample data source  （ Project code cloud download ）

Flink System examples And Build development environment and data

Sample module  （pom.xml）

Flink System examples And TableAPI & SQL And Sample module

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  Use Tbale&SQL And Flink Elasticsearch The connector writes data to kafka Message queue of 
 */
public class InsertToKafka {

    /**
         Official reference ：https://ci.apache.org/projects/flink/flink-docs-release-1.12/zh/dev/table/connectors/kafka.html

        format: For deserialization and serialization Kafka Format of message . Supported formats include 'csv','json','avro','debezium-json' and '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 {
        // structure StreamExecutionEnvironment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // Default flow time mode 
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
        // structure EnvironmentSettings  And designate Blink Planner
        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        // structure StreamTableEnvironment
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, bsSettings);
        // register kafka Data dimension table 
        tEnv.executeSql(table_sql);

        // Time format processing , Refer to Alibaba document 
        //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')))";

        //  The first way ： Direct execution sql
//        TableResult tableResult = tEnv.executeSql(sql);

        // The second way ： Declare a collection of operations to perform sql
        StatementSet stmtSet = tEnv.createStatementSet();
        stmtSet.addInsertSql(sql);
        TableResult tableResult = stmtSet.execute();

        tableResult.print();
    }
}

Print the results

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