Flink SQL (III) connects to the external system system and JDBC

file system

​ Another very common type of external system is the file system （File System） 了 .Flink Provides a connection to the file system Connector , Support reading and writing data from local or distributed file systems . This connector is built into Flink Medium , So there is no need to introduce additional dependencies to use it .

​ Here is an example of connecting to a file system ：

CREATE TABLE MyTable (
 column_name1 INT,
 column_name2 STRING,
 ...
 part_name1 INT,
 part_name2 STRING
) PARTITIONED BY (part_name1, part_name2) WITH (
 'connector' = 'filesystem', --  Connector type 
 'path' = '...', --  File path 
 'format' = '...' --  File format 
)

​ Here it is WITH Before using PARTITIONED BY Partition the data . File system connector support Access to partition files .

JDBC

​ The relational data table itself is SQL Where it was first applied , Therefore, we also hope to read and write table data directly to the relational database .Flink Provided JDBC The connector can be connected through JDBC The driver （driver） To any relational type Database read and write data , such as MySQL、PostgreSQL、Derby etc. .

​ As TableSink When writing data to the database , The mode of operation depends on how the table is created DDL Whether the primary key is defined （primary key）. If there is a primary key , that JDBC The connector will be inserted with an update （Upsert） mode , You can send a message to the external database according to the specified key （key） Update （UPDATE） And delete （DELETE） operation , If there is no primary key defined , Then we will add （Append） Run in mode , Update and delete operations are not supported .

1. Introduce dependencies

   Want to be in Flink Used in program JDBC The connector , The following dependencies need to be introduced ：

   <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-jdbc_${scala.binary.version}</artifactId>
    <version>${flink.version}</version>
   </dependency>
   

   Besides , To connect to a specific database , We also use to introduce related drive dependencies , such as MySQL：

   <dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>5.1.38</version>
   </dependency>
   

2. establish JDBC surface

   establish JDBC The method of the table is the same as before Upsert Kafka Be the same in essentials while differing in minor points . Here is a concrete example ：

   --  Create a connection to  MySQL  Of   surface 
   CREATE TABLE MyTable (
    id BIGINT,
    name STRING,
    age INT,
    status BOOLEAN,
    PRIMARY KEY (id) NOT ENFORCED
   ) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://localhost:3306/mydatabase',
    'table-name' = 'users'
   );
   --  Put another table  T  Data written to  MyTable  In the table 
   INSERT INTO MyTable
   SELECT id, name, age, status FROM T;
   
   

   ​ Here we create the table DDL The primary key is defined in , So the data will be in Upsert Mode write to MySQL In the table ; And then MySQL The connection of , It's through WITH In Clause url Defined . Pay attention to writing MySQL The real table name in is users, and MyTable It's registered in Flink Tables in the table environment .