kettle循环实现对一张大表数据分多次小批量delete

Question

需要对一张表做归档操作，归档过程：先将源表数据插入归档表，然后再delete源表数据。
因为如果归档100万条数据，一次性delete 100万的话，可能会锁表，并且delete这么大的数据量，也会很慢。于是，我想分小批量归档，比如每次归档1000条，执行完本批次归档后，再计算表中记录条数是否大于0，如果大于0的话，继续下一个批次归档，所以，用一个循环实现，总任务如下图：

问题：执行了几十次个批次之后，报错：内存溢出，看了一下CPU和内存都暴涨，kettle进程占了很大。有什么好方法可以实现上面需求？

Answer 1

引自免费微信小程序：皆我百晓生

要解决这个问题，我们可以采取一种更有效的方法来处理大数据的删除操作。Kettle本身提供了批处理的功能，但是当数据量特别大时，可能仍然会遇到性能瓶颈。这里推荐使用SQL的分区删除策略以及Kettle的流式转换（Stream Lookup）来优化这个流程。这样可以避免一次性加载大量数据到内存，从而减少内存消耗。

首先，你需要在源表上创建分区。例如，如果你的表有一个日期字段，你可以基于这个日期字段进行分区。这将有助于数据库管理系统更有效地处理大规模的数据操作。

1. 创建分区表:
   • 假设你的源表名为source_table，并且有一个名为date_field的日期字段。
   • 在MySQL中，创建分区表的命令可能是这样的：

CREATE TABLE source_table (
    -- 列定义...
) PARTITION BY RANGE (year(date_field)) (
    PARTITION p0 VALUES LESS THAN (2023),
    PARTITION p1 VALUES LESS THAN (2024)
    -- 添加更多分区，根据需要...
);

1. 优化删除操作:

   • 使用Kettle的Table Input步骤读取源表的ID或其他唯一标识符，而不是所有数据。
   • 创建一个Stream Lookup步骤，连接到归档表，查询已归档的ID。
   • 使用Filter Rows步骤过滤掉已经归档的ID。
   • 最后，使用Table Output步骤将剩余未归档的ID写入一个新的临时表，如temp_to_delete。

2. 分区删除:

   • 使用Table Input从temp_to_delete表读取ID。
   • 创建一个JavaScript步骤，在此步骤中编写一个脚本来动态构建分区范围内的删除语句。例如，如果当前处理的是2023年的数据，那么删除语句应该是针对p0分区的。
   • 使用Execute SQL Script步骤执行这些删除语句。

以下是大致的Kettle作业设计：

 Start -->
| Table Input (源表 ID) |
| Stream Lookup (归档表) |
| Filter Rows (排除已归档) |
| Table Output (temp_to_delete) |
|
V
+-----------------------+
| JavaScript            |   +-------------------+
|                       |   | Execute SQL Script|
| 动态构建删除语句... |   | 分区删除操作     |
+-----------------------+   +-------------------+
|
V
End

注意，JavaScript步骤中的代码取决于你使用的数据库类型，但基本思路是类似的。这是一个示例（使用JDBC）：

var connection = Spoon.createJobEntryCopy("jobentry_jdbc_connection", jobMeta).getDatabaseMeta().getConnection();
var dbType = connection.getDatabaseInterface().getName();

var sqlDelete = "";
var ids = new Array();

// 读取temp_to_delete表中的ID
var rows = Spoon rajobInputTableInput;
while (rows != null && rows.size() > 0) {
    var row = rows.get(0);
    ids.push(row.getString(1)); // 假设ID在第一列
    rows = spoon.nextRowSet(rows);
}

ids.forEach(function(id) {
    sqlDelete += "DELETE FROM source_table WHERE id = " + id + ";";
});

if (dbType == "mysql") {
    sqlDelete = "ALTER TABLE source_table AUTO_INCREMENT = 1; " + sqlDelete;
} else if (dbType == "postgresql") {
    // PostgreSQL不支持自动重置序列，因此需要单独处理
    // 这里省略PostgreSQL的具体代码，因为处理方式依赖于你的具体数据库结构
}

spoon.logBasic(sqlDelete);

try {
    var statement = connection.createStatement();
    statement.execute(sqlDelete);
} catch (e) {
    spoon.logError(e.getMessage());
}

确保在作业完成后清理临时表temp_to_delete。

通过这种方式，每次只处理一小部分数据，减少了内存占用，同时利用了数据库分区的优势，提高了删除效率。这种方法应该能显著降低内存消耗和提高整体性能。