flink读取数据，转化数据，写入数据
·
package cn.com.lyb.flink.source;
import com.ververica.cdc.connectors.postgres.PostgreSQLSource;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.connector.jdbc.*;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.util.BeanUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.HashMap;
import java.util.Map;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
public class PgSourceDemo {
// 源数据库配置（MySQL，读取 appointment 表）
private static final String SOURCE_JDBC_URL = "jdbc:mysql://172.32.x.x:3367/test2";
private static final String SOURCE_USERNAME = "root";
private static final String SOURCE_PASSWORD = "123456";
private static final String SOURCE_QUERY = "SELECT id, username, id_card, department, date, time, doctor_name " +
"FROM appointment " +
"WHERE id % ? = ?";
// 目标数据库配置（MySQL，写入 appointment1 表）
private static final String TARGET_JDBC_URL = "jdbc:mysql://172.32.x.x:3367/test2?rewriteBatchedStatements=true";
private static final String TARGET_USERNAME = "root";
private static final String TARGET_PASSWORD = "123456";
private static final String TARGET_TABLE = "appointment1";
private static final Logger log = LoggerFactory.getLogger(PgSourceDemo.class);
public static void main(String[] args) throws Exception {
long startTime = System.currentTimeMillis();
// 初始化Flink执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(10);
// 1. 从源表读取数据（Source）
env.addSource(new RichSourceFunction<Appointment>() {
private volatile boolean isRunning = true;
private Connection conn;
private PreparedStatement stmt;
private ResultSet rs;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
conn = DriverManager.getConnection(SOURCE_JDBC_URL, SOURCE_USERNAME, SOURCE_PASSWORD);
// 关键：获取当前子任务的「总并行度」和「子任务编号」
int parallelism = getRuntimeContext().getNumberOfParallelSubtasks(); // 总并行度（如10）
int subtaskIndex = getRuntimeContext().getIndexOfThisSubtask(); // 子任务编号（0~9）
// 设置SQL分区参数：id % 总并行度 = 子任务编号
stmt = conn.prepareStatement(SOURCE_QUERY);
stmt.setInt(1, parallelism); // 第一个?：总并行度
stmt.setInt(2, subtaskIndex); // 第二个?：子任务编号
}
@Override
public void run(SourceContext<Appointment> ctx) throws Exception {
rs = stmt.executeQuery();
while (isRunning && rs.next()) {
// 读取源表字段并封装为Appointment对象
int id = rs.getInt("id");
String username = rs.getString("username");
String idCard = rs.getString("id_card");
String department = rs.getString("department");
String date = rs.getString("date");
String time = rs.getString("time");
String doctorName = rs.getString("doctor_name");
// 发送到下游处理
ctx.collect(new Appointment(id, username, idCard, department, date, time, doctorName));
}
}
@Override
public void cancel() {
isRunning = false; // 标记任务停止
}
@Override
public void close() throws Exception {
// 关闭资源（确保连接释放）
super.close();
if (rs != null) rs.close();
if (stmt != null) stmt.close();
if (conn != null) conn.close();
}
})
.map(new MapFunction<Appointment, Appointment1>() {
@Override
public Appointment1 map(Appointment origin) throws Exception {
try {
if (origin.getId() < 1000) {
log.warn("不符合逻辑的数据");
return null;
}
// 原有数据处理逻辑不变
String processedUsername = "processed_" + origin.getUsername();
log.info("处理id===:{}", origin.getId());
return new Appointment1(
origin.getId(),
processedUsername,
origin.getIdCard(),
origin.getDepartment(),
origin.getDate(),
origin.getTime(),
origin.getDoctorName()
);
} catch (Exception e) {
log.error("处理数据id:{}时失败，返回null", origin.getId(), e);
return null;
}
}
})
// 新增Filter算子：直接过滤掉null对象，不进入后续Sink
.filter(processed -> {
boolean isNotNull = processed != null;
if (!isNotNull) {
log.warn("过滤掉null对象，不进入Sink");
}
return isNotNull;
})
// 3. 写入目标表（Sink）
.addSink(JdbcSink.sink(
// 插入SQL：字段与目标表appointment1保持一致
"INSERT INTO " + TARGET_TABLE +
" (id, username, id_card, department, date, time, doctor_name) " +
"VALUES (?, ?, ?, ?, ?, ?, ?) ",
// 绑定参数：按SQL中?的顺序设置
(statement, processed) -> {
if (processed == null) {
log.warn("跳过null对象，不执行插入操作");
return;
}
log.info("写入id---:{}", processed.getId());
// 插入部分参数（7个?）
statement.setInt(1, processed.getId());
statement.setString(2, processed.getUsername());
statement.setString(3, processed.getIdCard());
statement.setString(4, processed.getDepartment());
statement.setString(5, processed.getDate());
statement.setString(6, processed.getTime());
statement.setString(7, processed.getDoctorName());
},
// 目标库连接配置
// 批量执行策略（核心配置）
JdbcExecutionOptions.builder()
.withBatchSize(2000) // 累计2000条数据后批量提交
.withBatchIntervalMs(3000) // 若3秒内未达1000条，也批量提交（避免数据积压）
.withMaxRetries(3) // 失败重试3次
.build(),
new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
.withUrl(TARGET_JDBC_URL)
.withUsername(TARGET_USERNAME)
.withPassword(TARGET_PASSWORD)
.withDriverName("com.mysql.cj.jdbc.Driver")
.build()
));
// 执行任务
env.execute("MySQL to MySQL Data Processing");
long l = System.currentTimeMillis();
System.out.println(l - startTime);
}
}
class Appointment{
// 对应数据库字段：id（整数类型）
private int id;
// 对应数据库字段：username（用户名）
private String username;
// 对应数据库字段：id_card（身份证号）
private String idCard;
// 对应数据库字段：department（科室）
private String department;
// 对应数据库字段：date（预约日期）
private String date;
// 对应数据库字段：time（预约时间）
private String time;
// 对应数据库字段：doctor_name（医生姓名）
private String doctorName;
// 无参构造方法（Flink 序列化可能需要）
public Appointment() {
}
// 全参构造方法（用于初始化所有字段）
public Appointment(int id, String username, String idCard, String department, String date, String time, String doctorName) {
this.id = id;
this.username = username;
this.idCard = idCard;
this.department = department;
this.date = date;
this.time = time;
this.doctorName = doctorName;
}
// getter 和 setter 方法（用于访问和修改字段值）
public int getId() {
return id;
}
public void setId(int id) {
this.id = id;
}
public String getUsername() {
return username;
}
public void setUsername(String username) {
this.username = username;
}
public String getIdCard() {
return idCard;
}
public void setIdCard(String idCard) {
this.idCard = idCard;
}
public String getDepartment() {
return department;
}
public void setDepartment(String department) {
this.department = department;
}
public String getDate() {
return date;
}
public void setDate(String date) {
this.date = date;
}
public String getTime() {
return time;
}
public void setTime(String time) {
this.time = time;
}
public String getDoctorName() {
return doctorName;
}
public void setDoctorName(String doctorName) {
this.doctorName = doctorName;
}
}
class Appointment1{
// 对应数据库字段：id（整数类型）
private int id;
// 对应数据库字段：username（用户名）
private String username;
// 对应数据库字段：id_card（身份证号）
private String idCard;
// 对应数据库字段：department（科室）
private String department;
// 对应数据库字段：date（预约日期）
private String date;
// 对应数据库字段：time（预约时间）
private String time;
// 对应数据库字段：doctor_name（医生姓名）
private String doctorName;
// 无参构造方法（Flink 序列化可能需要）
public Appointment1() {
}
// 全参构造方法（用于初始化所有字段）
public Appointment1(int id, String username, String idCard, String department, String date, String time, String doctorName) {
this.id = id;
this.username = username;
this.idCard = idCard;
this.department = department;
this.date = date;
this.time = time;
this.doctorName = doctorName;
}
// getter 和 setter 方法（用于访问和修改字段值）
public int getId() {
return id;
}
public void setId(int id) {
this.id = id;
}
public String getUsername() {
return username;
}
public void setUsername(String username) {
this.username = username;
}
public String getIdCard() {
return idCard;
}
public void setIdCard(String idCard) {
this.idCard = idCard;
}
public String getDepartment() {
return department;
}
public void setDepartment(String department) {
this.department = department;
}
public String getDate() {
return date;
}
public void setDate(String date) {
this.date = date;
}
public String getTime() {
return time;
}
public void setTime(String time) {
this.time = time;
}
public String getDoctorName() {
return doctorName;
}
public void setDoctorName(String doctorName) {
this.doctorName = doctorName;
}
}
码道开发者社区，聚焦华为云码道 CodeArts 代码智能体，沉淀 Agent、Skill、鸿蒙开发实战内容，供开发者查阅资料、交流技术、分享工程实践
更多推荐
2
0- 0
已为社区贡献1条内容
所有评论(0)