|
|
@@ -9,7 +9,10 @@ import org.apache.flink.api.common.eventtime.WatermarkStrategy;
|
|
|
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
|
|
|
import org.apache.flink.api.common.time.Time;
|
|
|
import org.apache.flink.configuration.Configuration;
|
|
|
+import org.apache.flink.configuration.PipelineOptionsInternal;
|
|
|
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
|
|
|
+import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
|
|
|
+import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage;
|
|
|
import org.apache.flink.streaming.api.CheckpointingMode;
|
|
|
import org.apache.flink.streaming.api.datastream.DataStreamSource;
|
|
|
import org.apache.flink.streaming.api.environment.CheckpointConfig;
|
|
|
@@ -19,6 +22,7 @@ import org.springframework.beans.factory.annotation.Autowired;
|
|
|
import org.springframework.boot.CommandLineRunner;
|
|
|
import org.springframework.stereotype.Component;
|
|
|
|
|
|
+import javax.annotation.PreDestroy;
|
|
|
import java.io.File;
|
|
|
import java.util.*;
|
|
|
|
|
|
@@ -35,8 +39,8 @@ public class MySqlEventListener implements CommandLineRunner {
|
|
|
@Autowired
|
|
|
private FlinkConfig flinkConfig;
|
|
|
|
|
|
- private static final String JOB_ID = "mysql-cdc-sync-job";
|
|
|
- private static final String CHECKPOINT_DIR = "file:///c:/data/flink/checkpoints";
|
|
|
+ private static final String JOB_ID = "1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d";
|
|
|
+ private static final String CHECKPOINT_DIR = "d:/data/flink/checkpoints";
|
|
|
|
|
|
@Override
|
|
|
public void run(String... args) throws Exception {
|
|
|
@@ -66,9 +70,12 @@ public class MySqlEventListener implements CommandLineRunner {
|
|
|
streamSource.addSink(customSink)
|
|
|
.name("syncToRedis")
|
|
|
.setParallelism(flinkConfig.getParallelism().getSink());
|
|
|
+ // 修改执行方法，捕获更多信息
|
|
|
StreamGraph streamGraph = env.getStreamGraph();
|
|
|
- streamGraph.setJobName(JOB_ID);
|
|
|
-
|
|
|
+ streamGraph.setJobName("MySQL-CDC-Sync-Job");
|
|
|
+ // 执行前记录更多信息
|
|
|
+ log.info("准备执行CDC任务，作业名称: {}, 作业ID: {}, 检查点目录: {}",
|
|
|
+ streamGraph.getJobName(), JOB_ID, CHECKPOINT_DIR);
|
|
|
// 执行作业
|
|
|
env.execute(streamGraph);
|
|
|
}
|
|
|
@@ -105,11 +112,22 @@ public class MySqlEventListener implements CommandLineRunner {
|
|
|
}
|
|
|
|
|
|
private StreamExecutionEnvironment configureEnvironment() {
|
|
|
- final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
|
|
|
+ Configuration configuration = new Configuration();
|
|
|
+ configuration.setString(PipelineOptionsInternal.PIPELINE_FIXED_JOB_ID, JOB_ID);
|
|
|
+
|
|
|
+ // 添加这行 - 指定具体的检查点路径
|
|
|
+ String latestCheckpoint = findLatestCheckpoint();
|
|
|
+ if (latestCheckpoint != null) {
|
|
|
+ configuration.setString("execution.savepoint.path", latestCheckpoint);
|
|
|
+ log.info("将从检查点恢复: {}", latestCheckpoint);
|
|
|
+ }
|
|
|
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
|
|
|
|
|
|
// 设置状态后端
|
|
|
try {
|
|
|
- env.setStateBackend(new FsStateBackend(CHECKPOINT_DIR));
|
|
|
+ // 使用HashMap状态后端 (内存中计算，但检查点存储在文件系统)
|
|
|
+ env.setStateBackend(new HashMapStateBackend());
|
|
|
+ env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage(CHECKPOINT_DIR));
|
|
|
log.info("设置状态后端为: {}", CHECKPOINT_DIR);
|
|
|
} catch (Exception e) {
|
|
|
log.error("设置状态后端失败", e);
|
|
|
@@ -119,63 +137,68 @@ public class MySqlEventListener implements CommandLineRunner {
|
|
|
env.enableCheckpointing(flinkConfig.getCheckpoint().getInterval());
|
|
|
CheckpointConfig checkpointConfig = env.getCheckpointConfig();
|
|
|
checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
|
|
|
- checkpointConfig.setMinPauseBetweenCheckpoints(5000);
|
|
|
+ checkpointConfig.setMinPauseBetweenCheckpoints(500);
|
|
|
checkpointConfig.setCheckpointTimeout(60000);
|
|
|
checkpointConfig.setMaxConcurrentCheckpoints(1);
|
|
|
checkpointConfig.setTolerableCheckpointFailureNumber(3);
|
|
|
- checkpointConfig.enableExternalizedCheckpoints(
|
|
|
+ checkpointConfig.setExternalizedCheckpointCleanup(
|
|
|
CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
|
|
|
+
|
|
|
checkpointConfig.enableUnalignedCheckpoints();
|
|
|
|
|
|
// 设置重启策略
|
|
|
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.seconds(10)));
|
|
|
|
|
|
+
|
|
|
log.info("Flink环境配置完成，检查点间隔: {}ms, 作业ID: {}",
|
|
|
flinkConfig.getCheckpoint().getInterval(), JOB_ID);
|
|
|
|
|
|
return env;
|
|
|
}
|
|
|
|
|
|
- private StartupOptions getStartupOptions(String option) {
|
|
|
- if (option == null) {
|
|
|
- return StartupOptions.earliest();
|
|
|
- }
|
|
|
- switch (option.toLowerCase()) {
|
|
|
- case "initial": return StartupOptions.initial();
|
|
|
- case "latest": return StartupOptions.latest();
|
|
|
- case "latest-offset": return StartupOptions.latest(); // 显式处理 latest-offset
|
|
|
- default: return StartupOptions.latest();
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- private String findValidCheckpoint(String checkpointDirPath) {
|
|
|
- File checkpointDir = new File(checkpointDirPath);
|
|
|
+ private String findLatestCheckpoint() {
|
|
|
+ File checkpointDir = new File("d:/data/flink/checkpoints/" + JOB_ID);
|
|
|
if (!checkpointDir.exists() || !checkpointDir.isDirectory()) {
|
|
|
return null;
|
|
|
}
|
|
|
|
|
|
- // 查找完成的检查点目录 (通常格式为 chk-xx)
|
|
|
- File[] checkpointFiles = checkpointDir.listFiles(file ->
|
|
|
+ // 查找最新的检查点
|
|
|
+ File[] checkpoints = checkpointDir.listFiles(file ->
|
|
|
file.isDirectory() && file.getName().startsWith("chk-"));
|
|
|
|
|
|
- if (checkpointFiles == null || checkpointFiles.length == 0) {
|
|
|
- log.info("未找到有效的检查点");
|
|
|
+ if (checkpoints == null || checkpoints.length == 0) {
|
|
|
return null;
|
|
|
}
|
|
|
|
|
|
- // 按修改时间排序
|
|
|
- Arrays.sort(checkpointFiles, Comparator.comparingLong(File::lastModified).reversed());
|
|
|
-
|
|
|
- // 检查是否有 _metadata 文件，这通常表示完整的检查点
|
|
|
- for (File cpDir : checkpointFiles) {
|
|
|
- File metadataFile = new File(cpDir, "_metadata");
|
|
|
- if (metadataFile.exists()) {
|
|
|
- log.info("找到有效检查点: {}", cpDir.getAbsolutePath());
|
|
|
- return "file:///" + cpDir.getAbsolutePath();
|
|
|
- }
|
|
|
+ // 按检查点ID排序（提取数字部分）
|
|
|
+ Arrays.sort(checkpoints, (f1, f2) -> {
|
|
|
+ int id1 = Integer.parseInt(f1.getName().substring(4));
|
|
|
+ int id2 = Integer.parseInt(f2.getName().substring(4));
|
|
|
+ return Integer.compare(id2, id1); // 降序排列
|
|
|
+ });
|
|
|
+
|
|
|
+ // 检查_metadata文件是否存在
|
|
|
+ File latest = checkpoints[0];
|
|
|
+ File metadata = new File(latest, "_metadata");
|
|
|
+ if (metadata.exists() && metadata.isFile()) {
|
|
|
+ return "file:///d:/data/flink/checkpoints/" + JOB_ID + "/" + latest.getName();
|
|
|
}
|
|
|
|
|
|
- log.info("未找到包含元数据的有效检查点");
|
|
|
return null;
|
|
|
}
|
|
|
+
|
|
|
+ private StartupOptions getStartupOptions(String option) {
|
|
|
+ // 如果没有检查点，使用配置的启动模式
|
|
|
+ log.info("将使用配置的启动模式: {}", option);
|
|
|
+ if (option == null) {
|
|
|
+ return StartupOptions.earliest();
|
|
|
+ }
|
|
|
+ switch (option.toLowerCase()) {
|
|
|
+ case "initial": return StartupOptions.initial();
|
|
|
+ case "latest": return StartupOptions.latest();
|
|
|
+ case "earliest": return StartupOptions.earliest();
|
|
|
+ default: return StartupOptions.latest();
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
}
|
dongdong.xiang