# flink-cdc-sql

**Repository Path**: blucoding/flink-cdc-sql

## Basic Information

- **Project Name**: flink-cdc-sql
- **Description**: Flink CDC SQL 版实时同步系统
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-05-09
- **Last Updated**: 2026-05-09

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# Oracle-to-Oracle 实时同步系统 (Flink CDC SQL 版)

本工程是一个专为 Oracle 生产环境设计的高可用热备（Hot-Standby）同步方案。基于 Flink CDC 3.5.0 和 Flink 1.20.1 构建，采用动态元数据驱动的 SQL 同步引擎。

## 1. 核心特性
*   **自动化元数据解析**：启动时自动获取表字段、数据类型、精度。
*   **动态主键识别**：自动查询 Oracle 系统表获取主键约束，支持复合主键，确保 `Upsert` 语义。
*   **三种同步模式**：
    *   `LIST`：指定表名列表同步。
    *   `ALL`：全库同步 + 黑名单过滤。
    *   `REGEX`：基于正则表达式的表名筛选同步。
*   **生产级健壮性**：
    *   **断点续传**：基于 Flink Checkpoint 机制，支持停机重启后自动追溯 SCN。
    *   **高性能写入**：JDBC Sink 内置微批处理（Batching）与定时刷写逻辑。
    *   **幂等性保证**：主备库通过主键实现 `MERGE INTO` 逻辑，无惧数据重复。

## 2. 快速开始

### 2.1 依赖环境
*   **JDK**: 8 或 17
*   **Maven**: 3.6+
*   **Oracle**: 11g, 12c, 19c (需开启归档模式与补充日志)

### 2.2 Oracle 数据库准备 (重要)
在**主库**上以 `SYSDBA` 身份执行：
```sql
-- 1. 开启归档模式 (如果未开启，需重启数据库)
ALTER DATABASE ARCHIVELOG;

-- 2. 开启数据库级最小补充日志
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;

-- 3. 开启表级全列补充日志 (针对每张同步表执行)
ALTER TABLE yourdb.ACCEPT_INFO ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;

-- 4. 授予同步账号必要权限
GRANT SELECT ANY TABLE TO yourdb;
GRANT SELECT ANY DICTIONARY TO yourdb;
GRANT SELECT_CATALOG_ROLE TO yourdb;
GRANT LOGMINING TO yourdb;
```

### 2.3 修改配置
编辑 `src/main/resources/config.properties`：
*   配置主备库连接信息。
*   选择 `sync.mode` 并设置对应的过滤规则。
*   设置 `checkpoint.path` 用于存储同步进度。

### 2.4 编译与运行
**本地运行 (IDEA):**
1. 勾选运行配置中的 "Include dependencies with Provided scope"。
2. 直接运行 `OracleToOracleSQL.java`。

**打包运行 (集群):**
```bash
mvn clean package
./bin/flink run -d -c com.zxit.OracleToOracleSQL flink-cdc-sql-1.0-SNAPSHOT.jar
```

## 3. 同步模式说明

| 模式 | 配置项 | 说明 |
| :--- | :--- | :--- |
| **LIST** | `sync.tables` | 仅同步列表中的表，适合精确控制。 |
| **ALL** | `sync.exclude` | 同步 Schema 下所有表，排除黑名单中的临时/日志表。 |
| **REGEX** | `sync.pattern` | 通过正则表达式（如 `T_.*`）批量匹配业务表。 |

## 4. 生产运维手册

### 4.1 表结构变更 (DDL)
由于 SQL 模式暂不支持自动 DDL 传播，当主库修改结构时：
1. 停止 Flink 任务。
2. 在备库手动执行同样的 DDL。
3. 重启 Flink 任务（程序会自动解析新结构）。

### 4.2 灾难切换 (Failover)
当主库宕机需要切换至备库时：
1. **立即停止同步任务**（防止主库恢复后产生的脏数据污染备库）。
2. 将业务系统数据库连接指向备份库 IP。
3. 业务恢复。

### 4.3 重新全量同步
如果由于日志丢失或数据不一致需要重跑：
1. 停止任务。
2. **删除 `checkpoint.path` 指定的文件夹**。
3. 清空备库对应表的数据。
4. 重启任务。

## 5. 常见问题排查
*   **解析报错**：检查字段名是否包含特殊字符，当前代码已使用反引号（`）进行转义。
*   **延迟高**：观察 Oracle Redo Log 生成频率。在低频环境下，可增加 `ALTER SYSTEM SWITCH LOGFILE;` 测试。
*   **找不到类**：确保打包时使用了 `maven-shade-plugin` 且没排除必要的连接器依赖。
*   **ORA-14402: 更新分区关键字列将导致分区的更改**：目标数据库（Sink） 里的这张表是分区表，需要在目标数据库上对该表开启行移动（Row Movement）
* 

开启行移动（Row Movement）示例：
```sql
ALTER TABLE NJ120_0116.CALL_INFO ENABLE ROW MOVEMENT
```

查询需要开启行移动的所有分区表：
```sql
SELECT pt.table_name AS "未开启的分区表"
FROM user_part_tables pt
JOIN user_tables t ON pt.table_name = t.table_name
WHERE t.row_movement = 'DISABLED';
```