package edu.isi.karma.mapreduce.function;

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

public class CollectJSONObject extends UDF {
	private static Logger LOG = LoggerFactory.getLogger(MergeJSON.class);
	public Text evaluate(Text target, Text path) {
		try {
			if (path == null)
				return new Text("");
			String targetString = target.toString();
			String pathString = path.toString().replaceAll("\\[.*\\]","");
			JSONObject obj = new JSONObject(targetString);
			JSONArray array = new JSONArray();
			for (String aPath : pathString.split(",")) {
				String[] levels = aPath.split("\\.");
				collectJSONObject(obj, levels, 1, array);
			}
			return new Text(array.toString());
		}catch(Exception e) {
			LOG.error("something wrong",e );
			return new Text("");
		}
	}

	public void collectJSONObject(JSONObject obj, String[] levels, int i, JSONArray array) {
		if (i < levels.length && obj.has(levels[i])) {
			if (i == levels.length - 1) {
				Object value = obj.get(levels[i]);
				if (value instanceof JSONArray) {
					JSONArray t = (JSONArray)value;
					for (int j = 0; j < t.length(); j++) {
						array.put(t.get(j));
					}
				}
				else {
					array.put(value);
				}
			}
			else {
				Object value = obj.get(levels[i]);
				if (value instanceof JSONArray) {
					JSONArray t = (JSONArray)value;
					for (int j = 0; j < t.length(); j++) {
						try {
							JSONObject o = t.getJSONObject(j);
							collectJSONObject(o, levels, i + 1, array);
						}catch (Exception e) {

						}
					}
				}
				else if (value instanceof JSONObject){
					collectJSONObject((JSONObject)value, levels, i + 1, array);
				}

			}
		}
	}
}

@@ -36,16 +36,69 @@
 ction;%0A%0A
+import java.util.LinkedList;%0Aimport java.util.List;%0A%0A
 import o
@@ -751,45 +751,601 @@
 %09%09%09%09
+List%3C
 String
-%5B%5D
+%3E
  levels
- = aPath.split(%22%5C%5C.%22);
+Candidates = new LinkedList%3CString%3E();%0A%09%09%09%09int lastSplit = 0;%0A%09%09%09%09for(int i = 0; i %3C aPath.length() - 1; i++)%0A%09%09%09%09%7B%0A%09%09%09%09%09if(aPath.charAt(i) != '%5C%5C' && aPath.charAt(i+1) == '.')%0A%09%09%09%09%09%7B%0A%09%09%09%09%09%09levelsCandidates.add(aPath.substring(lastSplit, aPath.charAt(i) != '%5C%5C'? i+1 : i).replace(%22%5C%5C%22, %22%22));%0A%09%09%09%09%09%09lastSplit = i+2;%0A%09%09%09%09%09%7D%0A%09%09%09%09%7D%0A%09%09%09%09if(lastSplit %3C aPath.length())%0A%09%09%09%09%7B%0A%09%09%09%09%09levelsCandidates.add(aPath.substring(lastSplit, aPath.length()).replace(%22%5C%5C%22, %22%22));%0A%09%09%09%09%7D%0A%09%09%09%09String%5B%5D levels = new String%5BlevelsCandidates.size()%5D;%0A%09%09%09%09levels = levelsCandidates.toArray(levels);%0A%09%09%09%09
 %0A%09%09%09

Handle escaping periods in CollectJSONObject