任务：
将大量CSV文件写入数据库中；可以一张表一个csv文件输入，表输出；但是如果表文件太多，达到100多张表入库；那会写死人的；所以我们可以偷懒一下；采用配合kettle java脚本代码；完成任务；
场景：将下面文件写入数据中：文件名和表名对应关系如下：
文件名 表名
ccs_A_0101.csv ccs_A
ccs_B_0101.csv ccs_B
.......
文件格式：
字段1|+|字段2|+|字段3|+|字段4|+|.......|+|字段n
value1|+|Value2|+|value3|+||+|.........|+|valuen
实现：
1 新建一个kettle项目文件 ktr
涉及按钮：获取文件名，CSV文件输入，Java代码，SQL脚本
组合截图：
获取文件名：
CSV文件输入控件
java代码：后面单独放java代码块：
注意重写这两个函数：
/*设置参数变量*/
public void setVariable(String variableName, String variableValue) {
parent.setVariableImpl(variableName, variableValue);
}
/*获取参数变量*/
public String getVariable(String variableName) {
return parent.getVariableImpl(variableName);
}
执行SQL脚本：
涉及JAVA代码：
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Calendar;
/*设置参数变量*/
public void setVariable(String variableName, String variableValue) {
parent.setVariableImpl(variableName, variableValue);
}
/*获取参数变量*/
public String getVariable(String variableName) {
return parent.getVariableImpl(variableName);
}
/**修改参数变量：字段内容，表名*/
public String getTableFields(String field,String tablename){
if( "".equals(field) || field==null || "".equals(tablename)|| tablename==null ){
return null;
}
String[] valueArr = field.split("\\|\\+\\|",-1);//采用 -1模式，不去除末尾的空格
StringBuilder sb = new StringBuilder();
for(int i = 0; i< valueArr.length ;i++){
String value = valueArr[i];
sb.append("`").append(value).append("`,");
}
sb.deleteCharAt(sb.length() -1);
sb.append(",data_date,dw_created_time,dw_created_by,dw_updated_time,dw_updated_by");
return sb.toString(); //
}
/*设置表名或参数*/
public void setSelfVariables(String field,String tablename){
if( "".equals(field) || field==null || "".equals(tablename)|| tablename==null ){
return;
}
/**初始化参数*/
if(getVariable("TABLE_NAME") == null || "".equals(getVariable("TABLE_NAME"))){
setVariable("TABLE_NAME","");
setVariable("TABLE_FIELDS","");
setVariable("ROW_NUMBER","");
};
String tableName=getVariable("TABLE_NAME") ;//获取表名
/**表名不相同，就设置第一行，以及表名*/
if(!tableName.equals(tablename)){
setVariable("TABLE_NAME",tablename);
setVariable("TABLE_FIELDS",getTableFields(field,tablename));
setVariable("ROW_NUMBER","0");
}
else{
int row1=Integer.parseInt(getVariable("ROW_NUMBER")) ;
row1++;
setVariable("ROW_NUMBER",""+row1);
}
}
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());
// Get the value from an input field
String filename = get(Fields.In, "filename").getString(r);
if(filename.contains("/"))
filename=filename.substring(filename.lastIndexOf("/")+1,filename.indexOf("2") -1); //排除Linux环境下的：/tmp/ 反斜杠目录
if(filename.contains("\\"))
filename = filename.substring(filename.lastIndexOf("\\")+1,filename.indexOf("2") -1);//排除windows环境下的D:\ 斜杠目录
String tablename = filename;
tablename=tablename+"_detail";
String field = get(Fields.In,"field").getString(r);
/**如果表名相同，就退出 自定义函数*/
setSelfVariables(field,tablename); //自定义函数
String dataDate=getVariable("DATA_DATE","-02-23");// 外界传入参数：data_date 第二个参数为默认值：//String getVariable = getVariable(variableName, defaultValue);
String[] valueArr = field.split("\\|\\+\\|",-1);//采用 -1模式，不去除末尾的空格
StringBuilder sb = new StringBuilder();
for(int i = 0; i< valueArr.length ;i++){
String value = valueArr[i];
if(value == null || value.trim().length() == 0){
sb.append("null,");
}else{
sb.append("'");
sb.append(value.replace("'", "\\'"));
sb.append("',");
}
}
sb.deleteCharAt(sb.length() -1);
sb.append(",'"+dataDate+"'");
sb.append(",now(),'同步任务',null,null");
String resultStr = sb.toString();
//logDebug("tableName="+getVariable("TABLE_NAME")+" rowNumber="+ getVariable("ROW_NUMBER")+" resultStr = " + resultStr); //开发打印日志
//logDebug("tableFieldsStr="+getVariable("TABLE_FIELDS"));
get(Fields.Out, "tableName").setValue(r, getVariable("TABLE_NAME"));
get(Fields.Out, "tableFieldsStr").setValue(r, getVariable("TABLE_FIELDS"));
get(Fields.Out, "resultStr").setValue(r, resultStr);
get(Fields.Out, "rowNumber").setValue(r, getVariable("ROW_NUMBER"));
putRow(data.outputRowMeta, r);
return true;
}
