Flink CDC的事件时间处理_flink cdc 同步mysql 如何获取事件时间和处理时间

// 导入所需的包

import org.apache.flink.api.common.eventtime.WatermarkStrategy;

import org.apache.flink.streaming.api.datastream.DataStream;

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.api.bridge.java.TableEnvironment;

import org.apache.flink.table.api.TableResult;

import org.apache.flink.connector.jdbc.JdbcConnectionOptions;

import org.apache.flink.connector.jdbc.JdbcSink;

import org.apache.flink.connector.jdbc.JdbcStatementBuilder;

import org.apache.flink.streaming.api.functions.source.SourceFunction;

import org.apache.flink.streaming.api.functions.source.SourceContext;

import org.apache.flink.types.Row;

import org.apache.flink.util.Collector;

// 创建Flink执行环境

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();

StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);

// 创建CDC事件时间处理的数据流

DataStream<Row> cdcDataStream = env

.addSource(new MyCDCSourceFunction())

.returns(TypeInformation.of(Row.class))

.assignTimestampsAndWatermarks(WatermarkStrategy.<Row>forMonotonousTimestamps()

.withTimestampAssigner((event, timestamp) -> (long) event.getField("timestamp")));

// 将CDC数据流注册为表

tEnv.createTemporaryView("cdcTable", cdcDataStream, "id, timestamp, data");

// 执行SQL查询

TableResult result = tEnv.executeSql("SELECT id, COUNT(data) FROM cdcTable GROUP BY id");

// 将结果写入数据库

result.collect().forEach(row -> {

JdbcConnectionOptions connectionOptions = JdbcConnectionOptions.builder()

.withUrl("jdbc:mysql://localhost:3306/db")

.withDriverName("com.mysql.jdbc.Driver")

.withUsername("user")

.withPassword("password")

.build();

JdbcStatementBuilder<Row> statementBuilder = (ps, row) -> {

ps.setInt(1, (int) row.getField("id"));

ps.setLong(2, (long) row.getField(1));

};

cdcDataStream.addSink(JdbcSink.sink(

"INSERT INTO result_table (id, count) VALUES (?, ?)",

statementBuilder,

connectionOptions));

});

// 执行程序

env.execute("CDC Event Time Processing");