import java.io.IOException;
import java.util.*;

import org.apache.pig.EvalFunc;
import org.apache.pig.FuncSpec;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.DefaultBagFactory;
import org.apache.pig.data.DefaultTupleFactory;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.logicalLayer.schema.Schema;
import org.apache.pig.data.DataType;
import org.apache.pig.impl.logicalLayer.FrontendException;

public class Concat extends EvalFunc<DataBag> {
	
    @Override
    public DataBag exec(Tuple input) throws IOException {
    	
	if (input == null || input.size() == 0)
            return null;
        try{

        	if(input.get(0) != null && input.get(1) != null) {
	        	String token1 = input.get(0).toString();
	        	StringTokenizer st1 = new StringTokenizer(token1);
	        	String nodeId = st1.nextToken();
	        	
	        	String token2 = input.get(1).toString();
				StringTokenizer st2 = new StringTokenizer(token2);
				String platformId = st2.nextToken();
				
				
	        	DataBag output = DefaultBagFactory.getInstance().newDefaultBag();
				Tuple t = DefaultTupleFactory.getInstance().newTuple(1);
				String label = nodeId + ";" + platformId;
				t.set(0, label);
					
	            output.add(t);
	            return output;
        	}
        	
        	return null;
            
        }catch(Exception e){
            System.err.println("RealLabelsr: failed to process input; error - " + e.getMessage());
            e.printStackTrace();
            return null;
        }


    }
}