Tableapi & SQL and Kafka message acquisition of Flink example

Use Tbale&SQL And Flink Kafka Connector from kafka Get data from the 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

SelectToKafka.java

package com.flink.examples.kafka;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * @Description  Use Tbale&SQL And Flink Kafka Connector from kafka Get data from the message queue of 
 */
public class SelectToKafka {
    /**
      Official reference ：https://ci.apache.org/projects/flink/flink-docs-release-1.12/zh/dev/table/connectors/kafka.html
      Start offset position 
     config Options scan.startup.mode Appoint Kafka User's startup mode . Valid enumerations are ：
         group-offsets： From specific consumer groups ZK / Kafka The promise offset in the broker begins .
         earliest-offset： Start with the earliest offset .
         latest-offset： Start with the latest offset .
         timestamp： Start with the timestamp provided by the user of each partition .
         specific-offsets： Start with a specific offset provided by the user of each partition .
      Default option value group-offsets From ZK / Kafka The offset consumption last submitted in the broker 

      Consistency assurance 
     sink.semantic Option to select three different operation modes ：
         NONE：Flink Nothing can be guaranteed . The resulting records may be lost or may be repeated .
         AT_LEAST_ONCE （ default setting ）： This ensures that no records are lost （ Although they can repeat ）.
         EXACTLY_ONCE：Kafka Transactions will be used to provide a precise semantics . Whenever you use transaction write Kafka when , Please don't forget to use Kafka Record the settings required for any application settings isolation.level（read_committed  or read_uncommitted- The latter is the default ）.
     */
    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);

        String sql = "select user_id,item_id,behavior,ts from KafkaTable";
        Table table = tEnv.sqlQuery(sql);
        // Print the field structure 
        table.printSchema();

        //table  Turn into  dataStream  flow 
        DataStream<Row> behaviorStream = tEnv.toAppendStream(table, Row.class);
        behaviorStream.print();

        env.execute();
    }
}

Print the results

root
 |-- user_id: BIGINT
 |-- item_id: BIGINT
 |-- behavior: STRING
 |-- ts: TIMESTAMP(3)

3> 1,1,normal,2021-01-26T10:25:44