import cascading.operation.FunctionCall; //导入依赖的package包/类
@Override
public void operate(@SuppressWarnings("rawtypes") FlowProcess flowProcess, FunctionCall<Tuple> functionCall) {
    TupleEntry tupleEntry = functionCall.getArguments();
    Tuple outputTuple = functionCall.getContext();
    boolean allGroupFieldsValuesAreNull = true;
    for(Fields fields : groupFieldsPerPipe) {
        if(! groupFieldsValuesAreNull(tupleEntry, fields)) {
            for(int i=0; i<fields.size(); i++) {
                outputTuple.set(i, tupleEntry.getObject(fields.get(i)));
            }
            allGroupFieldsValuesAreNull = false;
            break;
        }
    }
    if(allGroupFieldsValuesAreNull) {
        outputTuple = Tuple.size(numOfGroupFields);
    }
    functionCall.getOutputCollector().add(outputTuple);
}
 

import cascading.operation.FunctionCall; //导入依赖的package包/类
@SuppressWarnings("rawtypes")
public void operate(FlowProcess process, FunctionCall call) {
	Data data = ((Data) call.getArguments().getObject(0)).deepCopy();
	DataUnit du = data.get_dataunit();
	if (du.getSetField() == DataUnit._Fields.PAGEVIEW) {
		String url = du.get_pageview().get_page().get_url();
		int time = data.get_pedigree().get_true_as_of_secs();
		call.getOutputCollector().add(new Tuple(url, time));
	}
}
 

import cascading.operation.FunctionCall; //导入依赖的package包/类
@SuppressWarnings("rawtypes")
public void operate(FlowProcess flowProcess, FunctionCall fnCall) {
	String sentence = fnCall.getArguments().getString(0);
	for (String word : sentence.split(" ")) {
		fnCall.getOutputCollector().add(new Tuple(word));
	}
}
 

import cascading.operation.FunctionCall; //导入依赖的package包/类
@Override
public void operate(FlowProcess flowProcess, FunctionCall functionCall) {
  TupleEntry arguments = functionCall.getArguments();
  Tuple result = new Tuple();

  Name name = new Name();
  name.setFirst_name(arguments.getString(0));
  name.setLast_name(arguments.getString(1));

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

import cascading.operation.FunctionCall; //导入依赖的package包/类
@Override
public void operate(FlowProcess flowProcess, FunctionCall functionCall) {
  Object record = functionCall.getArguments().getObject(0);
  Tuple result = new Tuple();
  result.add(record.toString());
  functionCall.getOutputCollector().add(result);
}
 

import cascading.operation.FunctionCall; //导入依赖的package包/类
@Override
public void operate(FlowProcess flowProcess, FunctionCall functionCall) {
  TupleEntry arguments = functionCall.getArguments();
  Tuple result = new Tuple();

  Tuple name = new Tuple();
  name.addString(arguments.getString(0));
  name.addString(arguments.getString(1));

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

import cascading.operation.FunctionCall; //导入依赖的package包/类
public void operate( FlowProcess flowProcess, FunctionCall functionCall )
{
TupleEntry argument = functionCall.getArguments();
String token = scrubText( argument.getString( "token" ) );

if( token.length() > 0 )
  {
  Tuple result = new Tuple();
  result.add( token );
  functionCall.getOutputCollector().add( result );
  }
}
 

import cascading.operation.FunctionCall; //导入依赖的package包/类
public void operate( FlowProcess flowProcess, FunctionCall functionCall )
{
TupleEntry argument = functionCall.getArguments();
String doc_id = argument.getString( 0 );
String token = scrubText( argument.getString( 1 ) );

if( token.length() > 0 )
  {
  Tuple result = new Tuple();
  result.add( doc_id );
  result.add( token );
  functionCall.getOutputCollector().add( result );
  }
}
 

import cascading.operation.FunctionCall; //导入依赖的package包/类
@SuppressWarnings("rawtypes")
public void operate(FlowProcess process, FunctionCall call) {
    System.out.println("DEBUG: " + call.getArguments().toString());
    call.getOutputCollector().add(new Tuple(1));
}
 

import cascading.operation.FunctionCall; //导入依赖的package包/类
@Override
public void operate(FlowProcess flowProcess, FunctionCall<Object> functionCall){
	Evaluator evaluator = getEvaluator();

	TupleEntry input = functionCall.getArguments();

	Map<FieldName, FieldValue> arguments = decodeArguments(evaluator, input);

	Map<FieldName, ?> result = evaluator.evaluate(arguments);

	TupleEntry output = encodeResult(evaluator, getFieldDeclaration(), result);

	TupleEntryCollector outputCollector = functionCall.getOutputCollector();

	outputCollector.add(output);
}