Kettle中通过Java对日期变量进行转换实现数据循环抽取

在对数据库中大量数据进入导出或抽取时，往往由于数据量太大不能一次全部导出或抽取出来，只能分多次运行导出，且每次运行抽取不同的数据。因此我们可以通过定义变量并在每次运行完后对变量进行转换(如日期变量可以增加相应的天数得到新的日期)，然后对变量进行判断，如果没有达到停止的条件则循环运行，如果达到停止的条件则中止任务。

比如我们需要将数据库中的某个表(ADT.INPATIENT)的2024年06月19日到2024年07月01日前的数据抽取到另外一张表(ADT.TEST11)，由于期间的数据量过大，不能一次全部抽取完，因此设定每次循环运行只能抽取2天的数据，直至将2024年06月31日的数据抽取完毕。

整个作业流程为：初始变量→检验字段的值→执行业务逻辑转换(带日期变量的业务转换)→日期变量累加n(对日期变量进行递增的转换)→检验字段的值.... 日期变量累加n(对日期变量进行递增的转换)→检验字段的值→中止作业(不满足循环执行条件)，如图：

第一步：创建【业务逻辑】转换：

打开Spoon.bat，在弹出的Kettle界面中依次选择【文件】/【新建】/【转换】，创建一个转换,保存名称为“业务逻辑转换”，并添加以下组件：

1.【表输入】：在左侧【核心对象】中检索“表输入”组件，双击该组件并在弹出的窗口中选择数据库链接、编写带有start_time和end_time变量的SQL语句(这2个变量将在作业中定义)，如图:

2.【插入/更新】：在左侧【核心对象】中选择“插入/更新”组件，选择“表输入”组件并按住【Shift】键,将两个组件连接起来。双击该组件并在弹出的窗口中选择数据库链接、目标模式(用户)、目标表、查询的关键字、更新的字段等，如图:

第二步：创建【日期变量累加n】转换：

打开Spoon.bat，在弹出的Kettle界面中依次选择【文件】/【新建】/【转换】，创建一个转换,保存名称为“日期变量累加n”

1.【获取变量】：在左侧【核心对象】中检索“获取变量”组件，双击该组件并在弹出的窗口中填写的变量start_time、end_time和days(该变量将在作业中进行定义)，如图:

2.【日期转换】：在左侧【核心对象】中检索“Java 代码”组件，双击该组件并在弹出的窗口中编写日期变量转换逻辑，如图:

说明：Java代码组件获取变量有3种方式：

方式一：通过getVariable方法获取变量：String varValue = System.getenv("myVariable");

方式二：通过环境变量获取变量：String varValue = System.getenv("myVariable");

方式三：通过参数获取变量：String varValue = getParameter("myParameter");

详细代码如下：

import org.pentaho.di.core.database.*;
import java.util.Date;
import java.text.*;
import java.util.*;

private DatabaseMeta databaseMeta;
private Database database;

public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException
{
	if (first){
	  first = false;
	}

	Object[] r = getRow();

	if (r == null) {
	  setOutputDone();
	  return false;
	}

    r = createOutputRow(r, data.outputRowMeta.size());

	// 获取时间步长变量
	String days = getVariable("days");
    if (days == null || days.trim() == "") {
        days = "0";
    }
	logBasic("获取时间步长====>"+days);

	// 获取开始时间.
    String startTime = getVariable("start_time");
	logBasic("转换前开始时间====>"+startTime);
    startTime = getDayStartTime(startTime,Integer.parseInt(days));
	
	// 获取endTime.
	String endTime = getVariable("end_time");
	logBasic("转换前结束时间====>"+endTime);
    endTime = getDayEndTime(endTime,Integer.parseInt(days));

	logBasic("转换后开始时间====>"+startTime);
	logBasic("转换后结束时间====>"+endTime);


	get(Fields.Out, "starttime").setValue(r, startTime);
	get(Fields.Out, "endtime").setValue(r, endTime);

    putRow(data.outputRowMeta, r);

	return true;
}


/**
 * 得到指定日期前后若干天日期的开始时间
 * @param date 日期字符串yyyymmdd或yyyyMMddHHmmss
 * @param days 加的天数
 * @return 新日期 yyyyMMddHHmmss
 */
public static String getDayStartTime(String date, int days) {
	java.util.Date ds;
	try {
		SimpleDateFormat sdf;
		SimpleDateFormat dft;
		if (date != null && date.trim().length() == 8) {
			sdf = new SimpleDateFormat("yyyyMMdd");
			dft = new SimpleDateFormat("yyyyMMdd");
		} else {
			sdf = new SimpleDateFormat("yyyyMMddHHmmss");
			dft = new SimpleDateFormat("yyyyMMddHHmmss");
		}

		ds = sdf.parse(date);
		Calendar calendar = Calendar.getInstance();
		calendar.setTime(ds);
		calendar.add(calendar.DAY_OF_YEAR, days);
		return (date != null && date.trim().length() == 8) ? dft.format(calendar.getTime())+"000000" : dft.format(calendar.getTime());
	} catch (ParseException e) {
		// TODO Auto-generated catch block
		e.printStackTrace();
	}
	return "";
}

/**
 * 得到指定日期前后若干天日期的结束时间
 * @param date 日期字符串yyyymmdd或yyyyMMddHHmmss
 * @param days 加的天数
 * @return 新日期 yyyyMMddHHmmss
 */
public static String getDayEndTime(String date, int days) {
	java.util.Date ds;
	try {
		SimpleDateFormat sdf;
		SimpleDateFormat dft;
		if (date != null && date.trim().length() == 8) {
			sdf = new SimpleDateFormat("yyyyMMdd");
			dft = new SimpleDateFormat("yyyyMMdd");
		} else {
			sdf = new SimpleDateFormat("yyyyMMddHHmmss");
			dft = new SimpleDateFormat("yyyyMMddHHmmss");
		}

		ds = sdf.parse(date);
		Calendar calendar = Calendar.getInstance();
		calendar.setTime(ds);
		calendar.add(calendar.DAY_OF_YEAR, days);
		return (date != null && date.trim().length() == 8) ? dft.format(calendar.getTime())+"235959" : dft.format(calendar.getTime());
	} catch (ParseException e) {
		// TODO Auto-generated catch block
		e.printStackTrace();
	}
	return "";
}

3.【设置变量】：在左侧【核心对象】中检索“设置变量”组件，双击该组件并在弹出的窗口中填写变量，如图:

第三步：创建作业：

打开Spoon.bat，在弹出的Kettle界面中依次选择【文件】/【新建】/【作业】，创建一个作业,保存名称为“变量日期自增作业”，并添加以下组件：

1.【启动】：在左侧【主对象树】中检索“Start”组件，双击该组件并在弹出的窗口中设置作业的调度频率，如图:

2.【初始变量】：在左侧【核心对象】中检索“设置变量”组件，双击该组件并在弹出的窗口中设置作业的变量名称、变量值，如图:

3.【校验字段的值】：在左侧【核心对象】中检索“检验字段的值”组件，双击该组件并在弹出的窗口中设置作业循环的判断条件，如图:

4【业务逻辑】：在左侧【核心对象】中检索“转换”组件，双击该组件并在弹出的窗口中设置作业对应的业务转换(数据的插入与抽取)，如图:

5.【日期变量累加n】：在左侧【核心对象】中检索“转换”组件，双击该组件并在弹出的窗口中设置作业日期变量累加转换，如图:

6.【中止作业】：在左侧【核心对象】中检索“中止作业”组件，当变量不符合循环条件时执行该步骤，如图:

第四步：启动作业：

点击作业左侧的【启动】按钮执行作业，如图: