参考社区实现 Flink1.11将聚合，update数据写入到kafka

我不是女神ヾ 2022-11-27 03:30 203阅读 0赞

1，快速链接社区文档：

[https://github.com/ververica/flink-cdc-connectors][https_github.com_ververica_flink-cdc-connectors]

下载依赖包：

>     <dependency>
>         <groupId>com.alibaba.ververica</groupId>
>         
>         <artifactId>flink-connector-mysql-cdc</artifactId>
>         <version>1.0.0</version>
>     </dependency>
>     <dependency>
>         <groupId>com.alibaba.ververica</groupId>
>         
>         <artifactId>flink-sql-connector-mysql-cdc</artifactId>
>         <version>1.0.0</version>
>     </dependency>
>     
>     <dependency>
>         <groupId>com.alibaba.ververica</groupId>
>         
>         <artifactId>flink-format-changelog-json</artifactId>
>         <version>1.0.0</version>
>     </dependency>
> 
>

我们观察到changelog- json 代码逻辑与canal-json一样 ：一个工厂类，一个反序列化，一个序列化类

![20200813155446760.png][]

上代码测试：

>     import org.apache.flink.streaming.api.TimeCharacteristic
>     import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
>     import org.apache.flink.table.api.EnvironmentSettings
>     import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
>     
>     object CdcSinkKafka {
>       def main(args: Array[String]): Unit = {
>         val env = StreamExecutionEnvironment.getExecutionEnvironment
>         env.enableCheckpointing(30001)
>         env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
>         val bsSettings = EnvironmentSettings.newInstance.useBlinkPlanner.inStreamingMode.build
>         val stenv = StreamTableEnvironment.create(env, bsSettings)
>     
>         val source =
>     
>           s"""
>              |CREATE TABLE kafka_table (
>              | category_id STRING,
>              | user_id INT,
>              | item_id STRING,
>              | behavior STRING,
>              | ts STRING
>              |) WITH (
>              | 'connector' = 'kafka',
>              | 'topic' = 'user_behavior',
>              | 'properties.bootstrap.servers' = 'dev-ct6-dc-worker01:9092,dev-ct6-dc-worker02:9092,dev-ct6-dc-worker03:9092',
>              | 'properties.group.id' = 'test1',
>              | 'format' = 'json',
>              | 'scan.startup.mode' = 'earliest-offset'
>              |)
>            """.stripMargin
>     
>         stenv.executeSql(source)
>     
>         val sink =
>           s"""
>              |CREATE TABLE kafka_gmv (
>              |  id STRING,
>              |  gmv DECIMAL(10, 5)
>              |) WITH (
>              |    'connector' = 'kafka',
>              |    'topic' = 'kafka_gmv',
>              |    'scan.startup.mode' = 'earliest-offset',
>              |    'properties.bootstrap.servers' = 'dev-ct6-dc-worker01:9092,dev-ct6-dc-worker02:9092,dev-ct6-dc-worker03:9092',
>              |    'format' = 'changelog-json'
>              |)
>            """.stripMargin
>     
>         stenv.executeSql(sink)
>         val insert =
>           s"""
>              | INSERT INTO kafka_gmv
>              |    SELECT behavior, SUM(user_id) as gmv
>              |    FROM kafka_table
>              |    GROUP BY behavior
>            """.stripMargin
>         stenv.executeSql(insert)
>         val query =
>           s"""
>              |SELECT * FROM kafka_gmv;
>            """.stripMargin
>     
>         stenv.executeSql(query).print()
>     
>     
>       }
>     }

kafka查看数据：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70][] \{"data":\{"id":"pv","gmv":955\},"op":"-U"\}

\{"data":\{"id":"pv","gmv":955\},"op":"+U"\}

[https_github.com_ververica_flink-cdc-connectors]: https://github.com/ververica/flink-cdc-connectors
[20200813155446760.png]: /images/20221124/532d8a0f51c14628850bf734b5a4419f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70]: /images/20221124/f8ebdf57fdb7495da53c31ee2b45b54d.png