package edu.isi.karma.mapreduce.function;

import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;

import org.apache.hadoop.hive.ql.exec.UDF;
import org.apache.hadoop.io.Text;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class SplitAndCleanJSONArray extends UDF {
	private static Logger LOG = LoggerFactory.getLogger(MergeJSON.class);
	public List<Text> evaluate(Text arrayToCleanText)
	{
		List<Text> cleanedValues = new LinkedList<Text>();
		if(arrayToCleanText == null)
		{
			return cleanedValues;
		}
		try{
			String arrayToClean = arrayToCleanText.toString();
			if(arrayToClean.startsWith("[") && arrayToClean.endsWith("]"))
			{
				arrayToClean = arrayToClean.substring(1,arrayToClean.length()-1);
			}
			
			Set<Text> cleanedValuesSet = new HashSet<Text>();
			
			String[] values = arrayToClean.split(",");
			for(String value : values)
			{
				cleanedValuesSet.add(new Text(value.replace("\"", "")));
			}
			
			cleanedValues.addAll(cleanedValuesSet);
		}
		catch(Exception e)
		{
			LOG.error("Unabled to split and clean array",e.getMessage());
		}
		return cleanedValues;
		
	}
}

@@ -178,16 +178,536 @@
 exec.UDF
+ArgumentException;%0Aimport org.apache.hadoop.hive.ql.exec.UDFArgumentLengthException;%0Aimport org.apache.hadoop.hive.ql.metadata.HiveException;%0Aimport org.apache.hadoop.hive.ql.udf.generic.GenericUDF;%0Aimport org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;%0Aimport org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorConverters;%0Aimport org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;%0Aimport org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory
 ;%0Aimport
@@ -836,16 +836,23 @@
 extends 
+Generic
 UDF %7B%0A%09p
@@ -926,59 +926,259 @@
 ;%0A%09p
-ublic List%3CText%3E evaluate(Text arrayToCleanText)%0A%09%7B
+rivate ObjectInspectorConverters.Converter%5B%5D converters;%0A%09@Override%0A%09public Object evaluate(DeferredObject%5B%5D arguments) throws HiveException%0A%09%7B%0A%09%09assert(arguments.length == 1);%0A%09%09Text arrayToCleanText = (Text)converters%5B0%5D.convert(arguments%5B0%5D.get());
 %0A%09%09L
@@ -1880,11 +1880,801 @@
 s;%0A%09%09%0A%09%7D
+%0A%0A%09@Override%0A%09public String getDisplayString(String%5B%5D arguments) %7B%0A%09%09assert (arguments.length == 1);%0A%09%09return %22SplitAndCleanJSONArray(%22 + arguments%5B0%5D + %22)%22;%0A%09%7D%0A%09@Override%0A%09public ObjectInspector initialize(ObjectInspector%5B%5D arguments)%0A%09%09%09throws UDFArgumentException %7B%0A%09%09if (arguments.length != 1) %7B%0A%09%09%09throw new UDFArgumentLengthException(%22The SplitAndCleanJSONArray takes only one argument%22);%0A%09%09%7D%0A%09%09converters = new ObjectInspectorConverters.Converter%5Barguments.length%5D;%0A%09%09for (int i = 0; i %3C arguments.length; i++) %7B%0A%09%09%09 converters%5Bi%5D = ObjectInspectorConverters.getConverter(arguments%5Bi%5D, PrimitiveObjectInspectorFactory.writableStringObjectInspector);%0A%09%09%7D%0A%09%09return ObjectInspectorFactory.getStandardListObjectInspector(PrimitiveObjectInspectorFactory.writableStringObjectInspector);%0A%09%7D
 %0A%7D%0A

change SplitAndCleanJSONArray to generic udf