創建新字段

Question

如何爲我的結果創建新字段？

我正在使用我想追加時間戳的wordcount作業。目前它包含字段'word'和'count'。

我的目標是創建一個元組，看起來像這樣： '字「數」「戳」

這是到目前爲止我的代碼。我嘗試添加自定義功能中的時間戳被稱爲 'TimestampAppender'

wcPipe = new Each(wcPipe, Fields.ALL, new TimestampAppender(Fields.ALL), Fields.RESULTS); 

TimestampAppender：

public class TimestampAppender extends BaseOperation implements Function { 
    public TimestampAppender(Fields fieldDeclaration) { 
    super(2, fieldDeclaration); 
} 

public void operate(FlowProcess flowProcess, FunctionCall functionCall) { 
    TupleEntry argument = functionCall.getArguments(); 

    String arg0 = argument.getString(0); 
    String arg1 = argument.getString(1); 

    Tuple result = new Tuple(); 
    result.addString(arg0); 
    result.addString(arg1); 
    result.addString("01-01-2015"); 

    functionCall.getOutputCollector().add(result); 
} 

Answer 1

你可以改變你的代碼爲：

wcPipe = new Each(wcPipe, new Insert(new Fields("timestamp"), ""), Fields.ALL); 
wcPipe = new Each(wcPipe, Fields.ALL, new TimestampAppender(Fields.ALL), Fields.RESULTS); 

TimestampAppender ：

public class TimestampAppender extends BaseOperation implements Function { 
    public TimestampAppender(Fields fieldDeclaration) { 
    super(Fields.ARGS); 
} 

public void operate(FlowProcess flowProcess, FunctionCall functionCall) { 
    TupleEntry argument = functionCall.getArguments(); 

    String arg0 = argument.getString(0); 
    String arg1 = argument.getString(1); 

    Tuple result = new Tuple(); 
    result.addString(arg0); 
    result.addString(arg1); 
    result.addString("01-01-2015"); 

    functionCall.getOutputCollector().add(result); 
}