kettle日志添加字段的源码修改

LogTableInterface.java(/kettle4.2/src-db/org/pentaho/di/core/logging/LogTableInterface.java)

73、74行添加接口方法：

//增加jobName参数 mengshanfeng
public RowMetaAndDatagetLogRecord(LogStatus status, Object subject, Object parent,String jobName);

保存后，会报如下几个实现此接口的实现类的错误，依次实现空方法即可。
ChannelLogTable(/kettle4.2/src/org/pentaho/di/core/logging/ChannelLogTable.java)
JobEntryLogTable(/kettle4.2/src/org/pentaho/di/core/logging/JobEntryLogTable.java)
JobLogTable(/kettle4.2/src/org/pentaho/di/core/logging/JobLogTable.java)
PerformanceLogTable(/kettle4.2/src/org/pentaho/di/core/logging/PerformanceLogTable.java)
StepLogTable(/kettle4.2/src/org/pentaho/di/core/logging/StepLogTable.java)
TransLogTable(/kettle4.2/src/org/pentaho/di/core/logging/TransLogTable.java)



TransLogTable.java(/kettle4.2/src/org/pentaho/di/core/logging/TransLogTable.java)

48、49行添加类变量

//mengshanfeng
public static String jobName = "";

330-374行注释掉

//并用如下代码代替：

return getLogRecord(status,subject,parent,TransLogTable.jobName);

378-433添加接口方法的实现

   

//增加jobName参数 mengshanfeng
public RowMetaAndData getLogRecord(LogStatus status, Object subject, Object parent,String jobName) {
//mengshanfeng
if(jobName!=null && !"".equals(jobName)){
TransLogTable.jobName = jobName;
}

if (subject==null || subject instanceof Trans) {
Trans trans = (Trans) subject;
Result result = null;
if (trans!=null) result = trans.getResult();

RowMetaAndData row = new RowMetaAndData();
int tt = 0;
for (LogTableField field : fields) {
if (field.isEnabled()) {
Object value = null;
if (trans!=null) {

switch(ID.valueOf(field.getId())){
case ID_BATCH : value = new Long(trans.getBatchId()); break;
case CHANNEL_ID : value = trans.getLogChannelId(); break;
case TRANSNAME : value = trans.getName(); break;
case STATUS : value = status.getStatus(); break;
case LINES_READ : value = new Long(result.getNrLinesRead()); break;
case LINES_WRITTEN : value = new Long(result.getNrLinesWritten()); break;
case LINES_INPUT : value = new Long(result.getNrLinesInput()); break;
case LINES_OUTPUT : value = new Long(result.getNrLinesOutput()); break;
case LINES_UPDATED : value = new Long(result.getNrLinesUpdated()); break;
case LINES_REJECTED : value = new Long(result.getNrLinesRejected()); break;
case ERRORS: value = new Long(result.getNrErrors()); break;
case STARTDATE: value = trans.getStartDate(); break;
case LOGDATE: value = trans.getLogDate(); break;
case ENDDATE: value = trans.getEndDate(); break;
case DEPDATE: value = trans.getDepDate(); break;
case REPLAYDATE: value = trans.getCurrentDate(); break;
case LOG_FIELD:
value = getLogBuffer(trans, trans.getLogChannelId(),status, logSizeLimit);
break;
}
}
tt = field.getDataType();
row.addValue(field.getFieldName(), field.getDataType(), value);
row.getRowMeta().getValueMeta(row.size()-1).setLength(field.getLength());
}
}

//增加两句话  mengshanfeng
row.addValue("jobName", tt, TransLogTable.jobName);
row.getRowMeta().getValueMeta(row.size()-1).setLength(10);
return row;
}
else {
return null;
}
}

 

Trans.java(/kettle4.2/src/org/pentaho/di/trans/Trans.java)

1692行添加函数调用参数

                

<pre name="code" class="java"> //增加jobName参数 mengshanfeng
transLogTableDatabaseConnection.writeLogRecord(transLogTable,LogStatus.START, this, null,parentJob.getJobname());

Database.java(/kettle4.2/src-db/org/pentaho/di/core/database/Database.java)

3621-3670行添加一个多态函数

    //增加同名函数，多一个jobName参数mengshanfeng
  

public void writeLogRecord(LogTableInterface logTable, LogStatus status, Object subject, Object parent,String jobName) throws KettleException {
try {
//增加jobName参数 mengshanfeng
RowMetaAndData logRecord = logTable.getLogRecord(status, subject, parent,jobName);
if (logRecord==null) return;

boolean update = (logTable.getKeyField()!=null) && !status.equals(LogStatus.START);
String schemaTable = databaseMeta.getQuotedSchemaTableCombination(
environmentSubstitute(logTable.getActualSchemaName()),
environmentSubstitute(logTable.getActualTableName())
);
RowMetaInterface rowMeta = logRecord.getRowMeta();
Object[] rowData = logRecord.getData();

if (update) {
RowMetaInterface updateRowMeta = new RowMeta();
Object[] updateRowData = new Object[rowMeta.size()];
ValueMetaInterface keyValueMeta = rowMeta.getValueMeta(0);
StringBuffer sqlBuff = new StringBuffer(250);
sqlBuff.append("UPDATE ").append( schemaTable ).append(" SET ");

for (int i = 1; i < rowMeta.size() ; i++) // Without ID_JOBor ID_BATCH
{
ValueMetaInterfacevalueMeta = rowMeta.getValueMeta(i);
if (i>1) {
sqlBuff.append(", ");
}
sqlBuff.append(databaseMeta.quoteField(valueMeta.getName())).append("=? ");

updateRowMeta.addValueMeta(valueMeta);
updateRowData[i-1] = rowData[i];
}
sqlBuff.append("WHERE").append(databaseMeta.quoteField(keyValueMeta.getName())).append("=? ");

updateRowMeta.addValueMeta(keyValueMeta);
updateRowData[rowMeta.size()-1] = rowData[0];

String sql = sqlBuff.toString();
execStatement(sql, updateRowMeta, updateRowData);

} else {

insertRow(environmentSubstitute(logTable.getActualSchemaName()),environmentSubstitute(logTable.getActualTableName()), logRecord.getRowMeta(), logRecord.getData());

}
} catch(Exception e) {
throw new KettleDatabaseException("Unable to write log record to log table " + logTable.getActualTableName(), e);
}
}

最终修改列表如下：

/kettle4.2/src-db/org/pentaho/di/core/logging/LogTableInterface.java

/kettle4.2/src-db/org/pentaho/di/core/database/Database.java

/kettle4.2/src/org/pentaho/di/core/logging/ChannelLogTable.java

/kettle4.2/src/org/pentaho/di/core/logging/JobEntryLogTable.java

/kettle4.2/src/org/pentaho/di/core/logging/JobLogTable.java

/kettle4.2/src/org/pentaho/di/core/logging/PerformanceLogTable.java

/kettle4.2/src/org/pentaho/di/core/logging/StepLogTable.java

/kettle4.2/src/org/pentaho/di/core/logging/TransLogTable.java

/kettle4.2/src/org/pentaho/di/trans/Trans.java

编译进Jar，进行测试即可。