Создать новое поле

Question

Как создать новое поле для моих результатов?

Я использую задачу wordcount, к которой я хочу добавить временную метку. В настоящее время он содержит поля «слово» и «счет».

Моя цель состоит в том, чтобы создать кортеж, который выглядит следующим образом: «слово» «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); 
}