实用工具之kettle分批作业

1.前言

通常如何数据库中的数据量非常的大，那么一次处理数据，可能就会导致kettle崩溃，不能一次性将全部的数据导入到数据表中。一方面是通过性能调优的方式，另外的一方面，是不是可以使用分页处理的方法进行分批处理数据呢？根据多方尝试和参考资料，我创建了可以进行分页处理的流程图，整个流程如下：

主要创建了一个分批作业，包括了：开始->获取页数->循环作业->成功，这几个部分。开始和成功不用说了，就是一个标志，主要是介绍获取页数和循环作业这两部分。

1.获取页数

获取页数，这是一个转换，可以新建一个转换，然后包括两个内容，一个是表输入，一个是复制记录到结果

1.1表输入

在表输入中，新建一个数据库连接，选择SQL Server,我的数据库是SQL Server数据库。

表输入的作用，主要就是产生一个临时表，表头的名字是P_PAGE，将全部的记录数，按每 5000 条 分成了 若干条记录，我的数据大约有45776295条记录，最后生成的表有9155条记录。

1

SELECT distinct P_PAGE from ( SELECT ROW_NUMBER() OVER (ORDER BY RecordTime)/5000 as P_PAGE FROM dbo.realtime ) as t order by P_PAGE

1.2复制记录到结果

复制记录到结果 很简单，就是从右侧的转换列表中，选择->作业->复制记录到结果，拖动到界面上，然后按住 Shift 键，进行连线就可以了。

2.循环作业

循环作业，是一个作业。为了对上一步产生的每一行执行一个操作，这里需要将作业属性中的 “执行每一个输入行” 勾选。

循环作业，包括：开始->设置变量->分页处理->成功，这几部分，开始和成功也不用说了，说一说设置变量和分页处理。

2.1设置变量

设置变量，是一个转换，包括两个组件，一个是 “从以前结果中获取记录”，一个是 “设置环境变量”。

(1) 从以前结果中获取记录
在右侧的转换列表中，选择->作业->“从结果中获取记录”,设置步骤名称，字段名称 P_PAGE,类型是Integer。

(2) 设置环境变量
在右侧的转换列表中，选择->作业->“设置变量”,设置步骤名称，字段名称 P_PAGE,变量名 VAR_P_PATE,变量活动类型选择 “Valid in the parant job”。

2.2分页处理

分页处理，又是一个转换流程，包括：表输入->字段选择->Java代码->表输出，四项。

(1) 表输入
从右侧选择输入->表输入，新建数据库连接，找到SQL Server数据库连接

SQL语句填入如下内容，因为我的SQL Server是2019,所以可以使用offset语句，我还勾选了 “替换SQL语句里面的变量”

1

select * from realtime order by RecordTime offset (${VAR_P_PAGE}*5000) row fetch next 5000 row only

(2) 字段选择
从右侧选择-> 转换-> 字段选择组件，在属性设置选择框中，将“选择和修改” 以及 “元数据” 中填入字段值，可以修改上一步的sql语句，为select * from table,先自动获取全部字段值，然后再修改上面的sql语句

(3) Java代码
从右侧选择->脚本->Java脚本 组件，我这里主要是处理了将 SiteNo这个字段的值全部统一加 200

输入java代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

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


	// First, get a row from the default input hop
	//
	Object[] r = getRow();

	// If the row object is null, we are done processing.
	//
	if (r == null) {
		setOutputDone();
		return false;
	}

	// It is always safest to call createOutputRow() to ensure that your output row's Object[] is large
	// enough to handle any new fields you are creating in this step.
	//
	Object[] outputRow = createOutputRow(r, data.outputRowMeta.size());
	// 刚开始我使用了Integer类型接收 getInteger() 函数返回的数值，总是不对
	// 因为我的数据库中的SittNo字段的值设置为了Integer(9),于是只能使用Long字段进行接收，
	// 这部分内容我测试了很久才终于搞清楚了这个格式
	Long siteno = get(Fields.In, "SiteNo").getInteger(r);

	// Set the value in the output field
	//
	Long temp = siteno+200L;
	get(Fields.Out, "SiteNo").setValue(outputRow, temp);

	// putRow will send the row on to the default output hop.
	//
	putRow(data.outputRowMeta, outputRow);

	return true;
}

(4) 表输出
从右侧选择->输出->表输出，新建 数据库连接，选择要插入数据的目标表，我这里忘记修改 提交记录数量，不知道有什么影响。

3.保存执行

依次创建好相关的步骤和转换之后，进行保存，直接回到开始的流程，点击执行就好了。我这里数据量比较大，电脑又太卡了，一直在执行中。

注意
我觉得可能还有一个地方需要注意的就是，要保存多次，或者是说如果有错误的时候，不一定是流程有问题，可能就是执行之前，没有保存或者是没有识别新的修改，我就遇到了这种情况，记得检查流程是不是设置的对。