使用 Flink DirectLoad 实现 OceanBase 旁路导入

# 下载 Flink（以 1.18.1 为例）
wget https://archive.apache.org/dist/flink/flink-1.19.3/flink-1.19.3-bin-scala_2.12.tgz

# 解压
tar -xzf flink-1.19.3-bin-scala_2.12.tgz

# 进入目录
cd flink-1.19.3

# 设置环境变量（可选但推荐）
export FLINK_HOME=$(pwd)

# 启动集群（包含 JobManager + TaskManager）
$FLINK_HOME/bin/start-cluster.sh

Starting cluster.
Starting standalonesession daemon on host your-hostname.
Starting taskexecutor daemon on host your-hostname.

# 查看 JobManager 日志
tail -f $FLINK_HOME/log/flink-*-standalonesession-*.log

# 查看 TaskManager 日志
tail -f $FLINK_HOME/log/flink-*-taskexecutor-*.log

# 示例：下载 1.5.0 版本
wget https://repo1.maven.org/maven2/com/oceanbase/flink-sql-connector-oceanbase-directload/1.0.0/flink-sql-connector-oceanbase-directload-1.5.0.jar

# 复制 JAR 到 Flink lib 目录
cp flink-sql-connector-oceanbase-directload-1.5.0.jar $FLINK_HOME/lib/

# 先停止
$FLINK_HOME/bin/stop-cluster.sh

# 再启动
$FLINK_HOME/bin/start-cluster.sh

obclient -h$host -P$port -u$user_name -p$password -D$database_name

# 测试 RPC 端口连通性
nc -zv <oceanbase-host> <rpc-port>

-- 连接到 OceanBase 数据库
USE test;

-- 创建目标表
CREATE TABLE `t_user` (
  `id` INT NOT NULL,
  `username` VARCHAR(50) DEFAULT NULL,
  `age` INT DEFAULT NULL,
  `score` DECIMAL(10,2) DEFAULT NULL,
  PRIMARY KEY (`id`)
) COMMENT '用户信息表';

# 启动 SQL Client
$FLINK_HOME/bin/sql-client.sh

▒▓██▓██▒
▓████▒▒█▓▒▓███▓▒
▓███▓░░        ▒▒▒▓██▒  ▒
░██▒   ▒▒▓▓█▓▓▒░      ▒████
██▒         ░▒▓███▒    ▒█▒█▒
░▓█            ███   ▓░▒██
▓█       ▒▒▒▒▒▓██▓░▒░▓▓█
█░ █   ▒▒░       ███▓▓█ ▒█▒▒▒
████░   ▒▓█▓      ██▒▒▒ ▓███▒
░▒█▓▓██       ▓█▒    ▓█▒▓██▓ ░█░
▓░▒▓████▒ ██         ▒█    █▓░▒█▒░▒█▒
███▓░██▓  ▓█           █   █▓ ▒▓█▓▓█▒
░██▓  ░█░            █  █▒ ▒█████▓▒ ██▓░▒
███░ ░ █░          ▓ ░█ █████▒░░    ░█░▓  ▓░
██▓█ ▒▒▓▒          ▓███████▓░       ▒█▒ ▒▓ ▓██▓
▒██▓ ▓█ █▓█       ░▒█████▓▓▒░         ██▒▒  █ ▒  ▓█▒
▓█▓  ▓█ ██▓ ░▓▓▓▓▓▓▓▒              ▒██▓           ░█▒
▓█    █ ▓███▓▒░              ░▓▓▓███▓          ░▒░ ▓█
██▓    ██▒    ░▒▓▓███▓▓▓▓▓██████▓▒            ▓███  █
▓███▒ ███   ░▓▓▒░░   ░▓████▓░                  ░▒▓▒  █▓
█▓▒▒▓▓██  ░▒▒░░░▒▒▒▒▓██▓░                            █▓
██ ▓░▒█   ▓▓▓▓▒░░  ▒█▓       ▒▓▓██▓    ▓▒          ▒▒▓
▓█▓ ▓▒█  █▓░  ░▒▓▓██▒            ░▓█▒   ▒▒▒░▒▒▓█████▒
██░ ▓█▒█▒  ▒▓▓▒  ▓█                █░      ░░░░   ░█▒
▓█   ▒█▓   ░     █░                ▒█              █▓
█▓   ██         █░                 ▓▓        ▒█▓▓▓▒█░
█▓ ░▓██░       ▓▒                  ▓█▓▒░░░▒▓█░    ▒█
██   ▓█▓░      ▒                    ░▒█▒██▒      ▓▓
▓█▒   ▒█▓▒░                         ▒▒ █▒█▓▒▒░░▒██
░██▒    ▒▓▓▒                     ▓██▓▒█▒ ░▓▓▓▓▒█▓
░▓██▒                          ▓░  ▒█▓█  ░░▒▒▒
▒▓▓▓▓▓▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒░░▓▓  ▓░▒█░

______ _ _       _       _____  ____  _         _____ _ _            _  BETA
|  ____| (_)     | |     / ____|/ __ \| |       / ____| (_)          | |
| |__  | |_ _ __ | | __ | (___ | |  | | |      | |    | |_  ___ _ __ | |_
|  __| | | | '_ \| |/ /  \___ \| |  | | |      | |    | | |/ _ \ '_ \| __|
| |    | | | | | |   <   ____) | |__| | |____  | |____| | |  __/ | | | |_
|_|    |_|_|_| |_|_|\_\ |_____/ \___\_\______|  \_____|_|_|\___|_| |_|\__|

Welcome! Enter 'HELP;' to list all available commands. 'QUIT;' to exit.

Flink SQL>

cd $FLINK_HOME

./bin/sql-client.sh

-- 可选：建议设置运行模式为 BATCH，可获得更好的性能
SET 'execution.runtime-mode' = 'BATCH';

-- 可选：设置并行度（根据数据量和集群资源调整并行度）
SET 'parallelism.default' = '4';

CREATE TABLE t_user (
  id INT,
  username STRING,
  age INT,
  score DECIMAL(10,2),
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'oceanbase-directload',
  'host' = 'xxx.x.x.x',                    -- OceanBase 主机地址
  'port' = 'xxxx',                         -- RPC 端口
  'tenant-name' = 'your_tenant-name',      -- 租户名
  'username' = 'your_username',            -- 用户名
  'password' = 'your_password',            -- 密码
  'schema-name' = 'test',                  -- 数据库名
  'table-name' = 't_user'                  -- 表名
);

-- 插入几条测试数据
INSERT INTO t_user
VALUES
  (1, 'Alice', 25, 95.5),
  (2, 'Bob', 30, 88.0),
  (3, 'Charlie', 28, 92.3),
  (4, 'David', 35, 87.8),
  (5, 'Eve', 22, 96.0);

[INFO] Submitting SQL update statement to the cluster...
[INFO] SQL update statement has been successfully submitted to the cluster:
Job ID: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx

SELECT * FROM test.t_user ORDER BY id;

+----+----------+------+-------+
| id | username | age  | score |
+----+----------+------+-------+
|  1 | Alice    |   25 | 95.50 |
|  2 | Bob      |   30 | 88.00 |
|  3 | Charlie  |   28 | 92.30 |
|  4 | David    |   35 | 87.80 |
|  5 | Eve      |   22 | 96.00 |
+----+----------+------+-------+
5 rows in set

# 停止 Flink 集群
cd $FLINK_HOME
./bin/stop-cluster.sh

-- 在 SQL Client 中执行
QUIT;

-- full 模式示例
CREATE TABLE t_sink (...) WITH (
  'connector' = 'oceanbase-directload',
  'load-method' = 'full',
  ...
);

-- inc 模式示例
CREATE TABLE t_sink (...) WITH (
  'connector' = 'oceanbase-directload',
  'load-method' = 'inc',
  'dup-action' = 'STOP_ON_DUP',  -- 不能使用 REPLACE
  ...
);

-- inc_replace 模式示例
CREATE TABLE t_sink (...) WITH (
  'connector' = 'oceanbase-directload',
  'load-method' = 'inc_replace',  -- dup-action 参数无效
  ...
);

'dup-action' = 'STOP_ON_DUP'

'dup-action' = 'REPLACE'

'dup-action' = 'IGNORE'

-- 示例：4C 租户建议设置
'parallel' = '8'  -- 或 '16'

-- 示例
'buffer-size' = '2048'

-- 1. 设置为 Batch 模式
SET 'execution.runtime-mode' = 'BATCH';

-- 2. 创建 Sink 表（使用最小参数集）
CREATE TABLE orders_sink (
  order_id BIGINT,
  user_id BIGINT,
  product_name STRING,
  amount DECIMAL(10, 2),
  order_time TIMESTAMP(3),
  PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
  'connector' = 'oceanbase-directload',
  'host' = '127.0.0.1',
  'port' = '2882',
  'tenant-name' = 'test',
  'username' = 'root',
  'password' = 'your_password',
  'schema-name' = 'test',
  'table-name' = 'orders'
);

-- 3. 从数据源导入（这里假设从另一个表读取）
INSERT INTO orders_sink
SELECT order_id, user_id, product_name, amount, order_time
FROM orders_source;

SET 'execution.runtime-mode' = 'BATCH';
SET 'parallelism.default' = '8';  -- 根据数据量调整

CREATE TABLE user_profile_sink (
  user_id BIGINT,
  username STRING,
  email STRING,
  age INT,
  city STRING,
  register_time TIMESTAMP(3),
  PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
  'connector' = 'oceanbase-directload',
  'host' = '127.0.0.1',
  'port' = '2882',
  'tenant-name' = 'prod',
  'username' = 'admin',
  'password' = 'your_password',
  'schema-name' = 'userdb',
  'table-name' = 'user_profile',
  -- 使用 full 模式（适合空表）
  'load-method' = 'full',
  -- 性能调优
  'parallel' = '16',  -- 服务端并行度
  'buffer-size' = '2048'
);

-- 从 CSV 文件或其他数据源导入
INSERT INTO user_profile_sink
SELECT * FROM user_profile_csv_source;

SET 'execution.runtime-mode' = 'BATCH';

CREATE TABLE product_info_sink (
  product_id BIGINT,
  product_name STRING,
  price DECIMAL(10, 2),
  stock INT,
  update_time TIMESTAMP(3),
  PRIMARY KEY (product_id) NOT ENFORCED
) WITH (
  'connector' = 'oceanbase-directload',
  'host' = '127.0.0.1',
  'port' = '2882',
  'tenant-name' = 'ecommerce',
  'username' = 'root',
  'password' = 'your_password',
  'schema-name' = 'productdb',
  'table-name' = 'product_info',
  -- 使用 inc_replace 模式（自动覆盖已存在的数据）
  'load-method' = 'inc_replace',  -- dup-action 参数会被忽略
  -- 性能配置
  'parallel' = '16',
  'buffer-size' = '2048'
);

-- 导入更新数据（自动覆盖旧记录）
INSERT INTO product_info_sink
SELECT * FROM product_info_latest;

SET 'execution.runtime-mode' = 'BATCH';

-- 创建 Kafka Source（有界模式）
CREATE TABLE kafka_source (
  event_id BIGINT,
  event_type STRING,
  user_id BIGINT,
  event_data STRING,
  event_time TIMESTAMP(3)
) WITH (
  'connector' = 'kafka',
  'topic' = 'user-events',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'flink-import-group',
  'scan.startup.mode' = 'earliest-offset',
  'scan.bounded.mode' = 'latest-offset',  -- 有界模式：读取到最新offset后停止
  'format' = 'json'
);

-- 创建 OceanBase Sink
CREATE TABLE event_sink (
  event_id BIGINT,
  event_type STRING,
  user_id BIGINT,
  event_data STRING,
  event_time TIMESTAMP(3),
  PRIMARY KEY (event_id) NOT ENFORCED
) WITH (
  'connector' = 'oceanbase-directload',
  'host' = '127.0.0.1',
  'port' = '2882',
  'tenant-name' = 'analytics',
  'username' = 'root',
  'password' = 'your_password',
  'schema-name' = 'eventdb',
  'table-name' = 'events',
  'load-method' = 'full',
  'parallel' = '16'
);

-- 导入数据
INSERT INTO event_sink
SELECT * FROM kafka_source;

SET 'execution.runtime-mode' = 'BATCH';

-- 创建文件 Source
CREATE TABLE csv_source (
  id BIGINT,
  name STRING,
  age INT,
  salary DECIMAL(10, 2)
) WITH (
  'connector' = 'filesystem',
  'path' = 'file:///data/employees.csv',
  'format' = 'csv',
  'csv.field-delimiter' = ',',
  'csv.ignore-parse-errors' = 'false'
);

-- 创建 OceanBase Sink
CREATE TABLE employee_sink (
  id BIGINT,
  name STRING,
  age INT,
  salary DECIMAL(10, 2),
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'oceanbase-directload',
  'host' = '192.168.1.100',
  'port' = '2882',
  'tenant-name' = 'hrms',
  'username' = 'admin',
  'password' = 'your_password',
  'schema-name' = 'hr',
  'table-name' = 'employees',
  'load-method' = 'full',
  'parallel' = '8',
  'buffer-size' = '2048',
  'max-error-rows' = '10'  -- 允许最多10行错误
);

-- 执行导入
INSERT INTO employee_sink
SELECT * FROM csv_source;

-- 根据数据量和集群资源设置并行度
SET 'parallelism.default' = '8';

-- 示例：8C 租户
'parallel' = '16'  -- 或 '32'

-- 小行数据（每行 < 1KB）
'buffer-size' = '4096'

-- 中等行数据（每行 1–10KB）
'buffer-size' = '1024'  -- 默认值

-- 大行数据（每行 > 10KB）
'buffer-size' = '512'

-- 设置2小时超时（根据数据量评估）
'timeout' = '2h'

-- 在 OceanBase 中手动触发合并
ALTER SYSTEM MAJOR FREEZE;

是否为首次导入到空表？
├─ 是 → 使用 full 模式（性能最优，列存效果最佳）
└─ 否（向非空表导入）
    └─ 是否需要覆盖已存在的记录？
        ├─ 是 → 使用 inc_replace 模式
        └─ 否 → 使用 inc 模式
            └─ 是否允许主键冲突？
                ├─ 否 → dup-action = STOP_ON_DUP
                ├─ 允许（保留新数据） → dup-action = REPLACE
                └─ 允许（保留旧数据） → dup-action = IGNORE