咨询根据日期将一行数据处理为多行数据的异常问题

--创建MSSQL源表

CREATE TABLE AA (

  id int NOT NULL  identity(1,1),

  username varchar(20) NULL,

  products varchar(100) NULL,

  LeaveBegin datetime,

  LeaveEnd datetime,

  LeaveHours numeric(19,2),

  LeaveDay numeric(19,2),

  PRIMARY KEY (id)

) 



--创建MSSQL目标表

CREATE TABLE BB (

  id int NOT NULL  identity(1,1),

  username varchar(20) NULL,

  products varchar(100) NULL,

  LeaveBegin datetime,

  LeaveEnd datetime,

  LeaveHours varchar(20) NULL,

  LeaveDay varchar(20) NULL,

  PRIMARY KEY (id)

) 



--创建模拟数据

insert  into AA(username,products,LeaveBegin,LeaveEnd,LeaveHours,LeaveDay) 

values ('name1','apple;milk;eggs;','2017-05-01','2017-05-02',16,2)

insert  into AA(username,products,LeaveBegin,LeaveEnd,LeaveHours,LeaveDay) 

values ('name2','potato,banana,milk;','2017-05-03','2017-05-06',32,4)

insert  into AA(username,products,LeaveBegin,LeaveEnd,LeaveHours,LeaveDay) 

values ('name3','shoes','2017-05-07','2017-06-11',288,36)

import java.text.SimpleDateFormat;

import java.util.Date;

import java.util.Calendar;

import java.io.UnsupportedEncodingException;



//kettle中已定义好的行处理方法，每行记录都会执行一次

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

{



	//(1)获取到上一个步骤的输入行

	Object[] r = getRow();

	if (r == null) {

	  setOutputDone();

	  return false;

	}

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



	//(2)读取出参数变量值

	String username= get(Fields.In,"username").getString(r);

	String products= get(Fields.In,"products").getString(r);

	String LeaveBegin = get(Fields.In, "LeaveBegin").getString(r);

	String LeaveEnd = get(Fields.In, "LeaveEnd").getString(r);

	String LeaveHours = get(Fields.In, "LeaveHours").getString(r);

	String LeaveDay = get(Fields.In, "LeaveDay").getString(r);



        //（3）处理数据	

	if(LeaveBegin!=null){		

		//获取请假开始到请假结束间隔多少天，精确到以天为单位

		SimpleDateFormat sdf=new SimpleDateFormat("yyyy-MM-dd");  

        Calendar cal = Calendar.getInstance();    

        long begintime = 0;

        long endtime = 0;

           

        try{

			cal.setTime(sdf.parse(LeaveBegin));   

            begintime = cal.getTimeInMillis();    

            cal.setTime(sdf.parse(LeaveEnd)); 

            endtime = cal.getTimeInMillis();  

           }catch(Exception e){

            e.printStackTrace();

          }

          long between_days=((endtime-begintime)/(1000*3600*24))+1;

		  

			//强制转换为INT类型

		    Float DateNum=Float.parseFloat(String.valueOf(between_days));

		  

		    //处理请假小时数

			Float TempLeaveHours=Float.parseFloat(LeaveHours)/DateNum;

			LeaveHours=String.valueOf(TempLeaveHours);	

			

			//处理请假天数

			Float TempLeaveDay=Float.parseFloat(LeaveDay)/DateNum;

			//Float TempLeaveDay=DateNum;

			LeaveDay=String.valueOf(TempLeaveDay);

		

		for(int i=0;i<DateNum;i++){

			Object[] clonedRow=getInputRowMeta().cloneRow(r);    //复制输入行数据

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

			

			//处理请假日期

			try{

				SimpleDateFormat ldf=new SimpleDateFormat("yyyy-MM-dd");

				Date LeaveNewDate = ldf.parse(LeaveBegin);

				Calendar dal = Calendar.getInstance(); 

				dal.setTime(LeaveNewDate); //注意此处的日期绑定字段

				dal.add(Calendar.DATE,i); //在请假日期的基础上加一天，直至请假结束日期

				

				//更新请假开始日期和请假结束日期

				LeaveBegin=ldf.format(dal.getTime());

				LeaveEnd=ldf.format(dal.getTime());

				

			}catch(Exception d){

            d.printStackTrace();

          }

			

			//(4)把处理好的值放入到输出记录中

			get(Fields.Out,"username").setValue(clonedRow,username);

			get(Fields.Out,"products").setValue(clonedRow,products);

			get(Fields.Out,"LeaveBegin").setValue(clonedRow,LeaveBegin);

			get(Fields.Out,"LeaveEnd").setValue(clonedRow,LeaveEnd);

			get(Fields.Out,"LeaveHours").setValue(clonedRow,LeaveHours);

			get(Fields.Out,"LeaveDay").setValue(clonedRow,LeaveDay);

			

			putRow(data.outputRowMeta,clonedRow);

		}



	}	

	

	//(5)输出到下一个节点做处理

   else putRow(data.outputRowMeta, r);

	return true;

}