依賴

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

2. 對(duì)應(yīng)數(shù)據(jù)庫(kù)的jdbc依賴，如以下為MySQL5.x的jdbc依賴
<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>5.1.49</version>
</dependency>

描述

• flinksql以多線程的方式將數(shù)據(jù)寫入外部數(shù)據(jù)表
• flinksql sink table的ddl中若聲明了主鍵，則上游數(shù)據(jù)到達(dá)sink table中將支持針對(duì)主鍵的append、update、delete模式（即upsert模式）
• mysql connector中，flinksql upsert模式通常需要和外部數(shù)據(jù)表的主鍵配合使用（即flink table和外部數(shù)據(jù)表的主鍵一致）

示例

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'" +
                ")" +
                "";