Flink SQL与JDBC的集成

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-jdbc_2.11</artifactId>
  <version>${flink.version}</version>
</dependency>
<dependency>
  <groupId>mysql</groupId>
  <artifactId>mysql-connector-java</artifactId>
  <version>5.1.38</version>
</dependency>

create table person(user_id  varchar(20), user_name  varchar(20), age int);
mysql> desc person;
+-----------+-------------+------+-----+---------+-------+
| Field     | Type        | Null | Key | Default | Extra |
+-----------+-------------+------+-----+---------+-------+
| user_id   | varchar(20) | YES  |     | NULL    |       |
| user_name | varchar(20) | YES  |     | NULL    |       |
| age       | int(11)     | YES  |     | NULL    |       |
+-----------+-------------+------+-----+---------+-------+

package com.aikfk.flink.sql.jdbc;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.catalog.hive.HiveCatalog;
public class FlinkKafkaJDBC {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().useBlinkPlanner().build();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env , settings);
        String catalogName = "flink_hive";
        String hiveDataBase = "flink";
        String hiveConfDir = "/Users/caizhengjie/Desktop/hive-conf";
        HiveCatalog hiveCatalog =
                new HiveCatalog(catalogName,hiveDataBase,hiveConfDir);
        tableEnvironment.registerCatalog(catalogName , hiveCatalog);
        tableEnvironment.useCatalog(catalogName);
        String kafkaTable = "kafka_person";
        String kafkaDropsql = "DROP TABLE IF EXISTS kafka_person";
        String kafakTable_sql
                = "CREATE TABLE kafka_person (\n" +
                "    user_id String,\n" +
                "    user_name String,\n" +
                "    age Int\n" +
                ") WITH (\n" +
                "   'connector.type' = 'kafka',\n" +
                "   'connector.version' = 'universal',\n" +
                "   'connector.topic' = 'kfk',\n" +
                "   'connector.properties.bootstrap.servers' = 'bigdata-pro-m07:9092',\n" +
                "   'format.type' = 'csv',\n" +
                "   'update-mode' = 'append'\n" +
                ")";
        tableEnvironment.executeSql(kafkaDropsql);
        tableEnvironment.executeSql(kafakTable_sql);
        // register a MySQL table 'person' in Flink SQL
        String mysqlTable_sql =
                "CREATE TABLE mysql_person (\n" +
                        "  user_id String,\n" +
                        "  user_name String,\n" +
                        "  age INT\n" +
                        ") WITH (\n" +
                        "   'connector' = 'jdbc',\n" +
                        "   'url' = 'jdbc:mysql://bigdata-pro-m07:3306/flink',\n" +
                        "   'table-name' = 'person',\n" +
                        "   'username' = 'root',\n" +
                        "   'password' = '199911'\n" +
                        ")";
        String mysqlDropsql = "DROP TABLE IF EXISTS mysql_person";
        tableEnvironment.executeSql(mysqlDropsql);
        tableEnvironment.executeSql(mysqlTable_sql);
        // write data into the JDBC table from the other table "kafka_person"
        tableEnvironment.executeSql("INSERT INTO mysql_person\n" +
                "SELECT user_id, user_name, age FROM kafka_person");
        env.execute("kafka");
    }
}

Flink SQL> show tables;
kafka_person
mysql_person
person
Flink SQL> desc kafka_person;
+-----------+--------+------+-----+--------+-----------+
|      name |   type | null | key | extras | watermark |
+-----------+--------+------+-----+--------+-----------+
|   user_id | STRING | true |     |        |           |
| user_name | STRING | true |     |        |           |
|       age |    INT | true |     |        |           |
+-----------+--------+------+-----+--------+-----------+
3 rows in set

>100,alex,10
>100,alex,10
>100,alex,10
>100,alex,10
>100,alex,10
>100,alex,10

mysql> select * from person;
+---------+-----------+------+
| user_id | user_name | age  |
+---------+-----------+------+
| 100     | alex      |   10 |
| 100     | alex      |   10 |
| 100     | alex      |   10 |
| 100     | alex      |   10 |
| 100     | alex      |   10 |
+---------+-----------+------+
5 rows in set (0.00 sec)

select * from kafka_person;
select * from mysql_person;