依赖

1. flink jdbc api
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-jdbc_2.12</artifactId>
    <version>${flink.version}</version>
</dependency>

2. 对应数据库的jdbc依赖，如以下为MySQL5.x的jdbc依赖
<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>5.1.49</version>
</dependency>

描述

• flinksql以多线程的方式将数据写入外部数据表
• flinksql sink table的ddl中若声明了主键，则上游数据到达sink table中将支持针对主键的append、update、delete模式（即upsert模式）
• mysql connector中，flinksql upsert模式通常需要和外部数据表的主键配合使用（即flink table和外部数据表的主键一致）

示例

String sinkDDL = "create table mysqlSink(" +
                "ts timestamp," +
                "user_id bigint," +
                "item_id bigint," +
                "behavior string" +
                ") with (" +
                "'connector'='jdbc'," +
                "'url'='jdbc:mysql://localhost:3306/test'," +
                "'table-name'='user_event'," +
                "'username'='root'," +
                "'password'='123456'" +
                ")" +
                "";