kettle9.0 多库多表同步数据

一、需求：有30个库，每个库有53张表，需要将表中数据完整的同步到另一个库中，做备份用

1、主作业：读取数据库表信息，循环每个库。

 

2、次作业：读取库中的所有表，循环每个表。

 

 

 

 

// 作业级读取变量：用 parent_job.getVariable()（Kettle 9.0 作业级兼容写法）
var syncType = parent_job.getVariable("SYNC_TYPE", "increment");

// 运行后看日志中是否能看到"当前SYNC_TYPE=xxx"，确认变量读取正确
if (syncType == "increment") {
// 增量：返回true → 走成功分支
true;
} else {
// 全量：返回false → 走失败分支
false;
}

 

3、单表作业：读取表中信息，根据数据的特点用来区分是全量同步还是增量同步。

 

 

public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException {

Object[] rowData = getRow();
if (rowData == null) {
setOutputDone();
return false;
}

// 读取转换级变量（适配Kettle 9.0）
String tableName = getTrans().getVariable("TABLE_NAME", "");
String fieldList = getTrans().getVariable("FIELD_LIST", "");
String pkField = "id";

// 空值校验：表名/字段列表为空时跳过
if (tableName.trim().isEmpty() || fieldList.trim().isEmpty()) {
logError("表名或字段列表为空，跳过当前行处理");
putRow(data.outputRowMeta, rowData);
return true;
}

// 拆分字段列表
String[] sourceFields = fieldList.trim().isEmpty() ? new String[0] : fieldList.split(",");

StringBuilder fields = new StringBuilder();
StringBuilder values = new StringBuilder();
StringBuilder updates = new StringBuilder();

// 遍历字段拼接SQL（仅处理源/目标表一致的字段）
for (int i = 0; i < sourceFields.length; i++) {
String fieldName = sourceFields[i].trim();
if (fieldName.isEmpty()) continue;

// 获取字段值和元数据
int idx = getInputRowMeta().indexOfValue(fieldName);
Object val = (idx >= 0) ? rowData[idx] : null;
ValueMetaInterface meta = (idx >= 0) ? getInputRowMeta().getValueMeta(idx) : null;

// 拼接字段名
fields.append("`").append(fieldName).append("`");

// 拼接字段值（兼容不同数据类型）
if (val == null) {
values.append("NULL");
} else if (meta != null && (meta.isNumeric() || meta.isInteger() || meta.isBigNumber())) {
values.append(val.toString());
} else if (meta != null && meta.isBoolean()) {
boolean b = "1".equals(val.toString()) || "true".equalsIgnoreCase(val.toString());
values.append(b ? "1" : "0");
} else {
String rawStr = val.toString();
String cleanStr = rawStr.replaceAll("[^\\u0000-\\uFFFF]", ""); // 移除 Emoji 等
String escapedStr = cleanStr.replace("\\", "\\\\").replace("'", "''");
values.append("'").append(escapedStr).append("'");
}

// 拼接更新子句（排除主键字段）
if (!fieldName.equalsIgnoreCase(pkField)) {
updates.append("`").append(fieldName).append("` = VALUES(`").append(fieldName).append("`)");
}

// 最后一个字段不加逗号，避免SQL语法错误
if (i < sourceFields.length - 1) {
fields.append(",");
values.append(",");
if (!fieldName.equalsIgnoreCase(pkField)) {
updates.append(",");
}
}
}

// 生成最终的INSERT/UPDATE SQL
String sql = "INSERT INTO `" + tableName + "` (" + fields.toString() + ") VALUES (" + values.toString() + ") ON DUPLICATE KEY UPDATE " + updates.toString();

// 输出SQL到下一个步骤
Object[] outputRow = createOutputRow(rowData, data.outputRowMeta.size());
get(Fields.Out, "dynamic_sql").setValue(outputRow, sql);
putRow(data.outputRowMeta, outputRow);

return true;
}

 全量同步的时候，需要查询所有的数据，在表输出的时候需要勾选裁剪表，这个代表插入数据之前先清空数据。