@Test public void testFlattenToDepth() throws Exception { String[] directives = new String[]{ "parse-as-json body 2" }; List<Row> rows = Arrays.asList( new Row("body", "{\"created_at\":\"Mon Feb 06 21:13:37 +0000 2017\",\"id\":828713281267367937," + "\"id_str\":\"828713281267367937\",\"text\":\"Youth is counted sweetest by those who are no longer young." + "\\n#PBBPADALUCKKISSYONG\",\"source\":\"\\u003ca href=\\\"http:\\/\\/twitter.com\\/download\\/iphone\\\" " + "rel=\\\"nofollow\\\"\\u003eTwitter for iPhone\\u003c\\/a\\u003e\",\"truncated\":false," + "\"in_reply_to_status_id\"" + ":null,\"in_reply_to_status_id_str\":null,\"in_reply_to_user_id\":null,\"in_reply_to_user_id_str\":null,\"in_" + "reply_to_screen_name\":null,\"user\":{\"id\":520417116,\"id_str\":\"520417116\",\"name\":\"glenda andres\"," + "\"screen_name\":\"glenda_andres\",\"location\":\"Israel\",\"url\":null," + "\"description\":\"volleyball,badminton\"," + "\"protected\":false,\"verified\":false,\"followers_count\":12,\"friends_count\":94,\"listed_count\":1," + "\"favourites_count\":42,\"statuses_count\":595,\"created_at\":\"Sat Mar 10 13:57:16 +0000 2012\"," + "\"utc_offset" + "\":null,\"time_zone\":null,\"geo_enabled\":false,\"lang\":\"en\",\"contributors_enabled\":false," + "\"is_translator" + "\":false,\"profile_background_color\":\"C0DEED\",\"profile_background_image_url\":\"http:\\/\\/" + "abs.twimg.com\\" + "/images\\/themes\\/theme1\\/bg.png\",\"profile_background_image_url_https\":\"https:\\/\\/abs.twimg.com\\/" + "images" + "\\/themes\\/theme1\\/bg.png\",\"profile_background_tile\":false,\"profile_link_color\":\"1DA1F2\",\"profile_" + "sidebar_border_color\":\"C0DEED\",\"profile_sidebar_fill_color\":\"DDEEF6\"," + "\"profile_text_color\":\"333333\"," + "\"profile_use_background_image\":true,\"profile_image_url\":\"http:\\/\\/pbs.twimg.com\\/profile_images\\" + "/680735812972052480\\/hsvuZASG_normal.jpg\",\"profile_image_url_https\":\"https:\\/\\/pbs.twimg.com\\/" + "profile_" + "images\\/680735812972052480\\/hsvuZASG_normal.jpg\",\"profile_banner_url\":\"https:\\/\\/pbs.twimg.com\\/" + "profile_" + "banners\\/520417116\\/1451511037\",\"default_profile\":true,\"default_profile_image\":false," + "\"following\":null," + "\"follow_request_sent\":null,\"notifications\":null},\"geo\":null,\"coordinates\":null,\"place\":null," + "\"contributors\":null,\"is_quote_status\":false,\"retweet_count\":0,\"favorite_count\":0,\"entities\":" + "{\"hashtags\":[{\"text\":\"PBBPADALUCKKISSYONG\",\"indices\":[60,80]}],\"urls\":[],\"user_mentions\":[]," + "\"symbols\":[]},\"favorited\":false,\"retweeted\":false,\"filter_level\":\"low\",\"lang\":\"en\"," + "\"timestamp_ms\":\"1486415617659\"}")); rows = TestingRig.execute(directives, rows); Assert.assertTrue(rows.size() == 1); }
@Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { List<Row> results = new ArrayList<>(); for (Row row : rows) { int idx = row.find(column); if (idx != -1) { Object value = row.getValue(idx); if (value == null) { continue; } try { JsonElement element = null; if (value instanceof String) { String document = (String) value; element = parser.parse(document.trim()); } else if (value instanceof JsonObject || value instanceof JsonArray) { element = (JsonElement) value; } else { throw new DirectiveExecutionException( NAME, String.format("Column '%s' is of invalid type '%s'. It should be of type 'String'" + " or 'JsonObject' or 'JsonArray'.", column, value.getClass().getSimpleName())); } row.remove(idx); if (element != null) { if (element instanceof JsonObject) { jsonFlatten(element.getAsJsonObject(), column, 1, depth, row); results.add(row); } else if (element instanceof JsonArray) { JsonArray array = element.getAsJsonArray(); if (array.size() > 0) { for (int i = 0; i < array.size(); ++i) { JsonElement object = array.get(i); Row newRow = new Row(row); newRow.add(column, getValue(object)); results.add(newRow); } } else { results.add(row); } } else if (element instanceof JsonPrimitive) { row.add(column, getValue(element.getAsJsonPrimitive())); } } } catch (JSONException e) { throw new ErrorRowException(NAME, e.getMessage(), 1); } } } return results; }
JsParser implements Directive, Lineage { @Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { List<Row> results = new ArrayList<>(); for (Row row : rows) { int idx = row.find(column); if (idx != -1) { Object value = row.getValue(idx); if (value == null) { continue; } try { JsonElement element = null; if (value instanceof String) { String document = (String) value; element = parser.parse(document.trim()); } else if (value instanceof JsonObject || value instanceof JsonArray) { element = (JsonElement) value; } else { throw new DirectiveExecutionException( NAME, String.format("Column '%s' is of invalid type '%s'. It should be of type 'String'" + " or 'JsonObject' or 'JsonArray'.", column, value.getClass().getSimpleName())); } row.remove(idx); if (element != null) { if (element instanceof JsonObject) { jsonFlatten(element.getAsJsonObject(), column, 1, depth, row); results.add(row); } else if (element instanceof JsonArray) { JsonArray array = element.getAsJsonArray(); if (array.size() > 0) { for (int i = 0; i < array.size(); ++i) { JsonElement object = array.get(i); Row newRow = new Row(row); newRow.add(column, getValue(object)); results.add(newRow); } } else { results.add(row); } } else if (element instanceof JsonPrimitive) { row.add(column, getValue(element.getAsJsonPrimitive())); } } } catch (JSONException e) { throw new ErrorRowException(NAME, e.getMessage(), 1); } } } return results; } }
JsParser implements Directive, Lineage { @Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { List<Row> results = new ArrayList<>(); for (Row row : rows) { int idx = row.find(column); if (idx != -1) { Object value = row.getValue(idx); if (value == null) { continue; } try { JsonElement element = null; if (value instanceof String) { String document = (String) value; element = parser.parse(document.trim()); } else if (value instanceof JsonObject || value instanceof JsonArray) { element = (JsonElement) value; } else { throw new DirectiveExecutionException( NAME, String.format("Column '%s' is of invalid type '%s'. It should be of type 'String'" + " or 'JsonObject' or 'JsonArray'.", column, value.getClass().getSimpleName())); } row.remove(idx); if (element != null) { if (element instanceof JsonObject) { jsonFlatten(element.getAsJsonObject(), column, 1, depth, row); results.add(row); } else if (element instanceof JsonArray) { JsonArray array = element.getAsJsonArray(); if (array.size() > 0) { for (int i = 0; i < array.size(); ++i) { JsonElement object = array.get(i); Row newRow = new Row(row); newRow.add(column, getValue(object)); results.add(newRow); } } else { results.add(row); } } else if (element instanceof JsonPrimitive) { row.add(column, getValue(element.getAsJsonPrimitive())); } } } catch (JSONException e) { throw new ErrorRowException(NAME, e.getMessage(), 1); } } } return results; }  }
JsParser implements Directive, Lineage { @Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { List<Row> results = new ArrayList<>(); for (Row row : rows) { int idx = row.find(column); if (idx != -1) { Object value = row.getValue(idx); if (value == null) { continue; } try { JsonElement element = null; if (value instanceof String) { String document = (String) value; element = parser.parse(document.trim()); } else if (value instanceof JsonObject || value instanceof JsonArray) { element = (JsonElement) value; } else { throw new DirectiveExecutionException( NAME, String.format("Column '%s' is of invalid type '%s'. It should be of type 'String'" + " or 'JsonObject' or 'JsonArray'.", column, value.getClass().getSimpleName())); } row.remove(idx); if (element != null) { if (element instanceof JsonObject) { jsonFlatten(element.getAsJsonObject(), column, 1, depth, row); results.add(row); } else if (element instanceof JsonArray) { JsonArray array = element.getAsJsonArray(); if (array.size() > 0) { for (int i = 0; i < array.size(); ++i) { JsonElement object = array.get(i); Row newRow = new Row(row); newRow.add(column, getValue(object)); results.add(newRow); } } else { results.add(row); } } else if (element instanceof JsonPrimitive) { row.add(column, getValue(element.getAsJsonPrimitive())); } } } catch (JSONException e) { throw new ErrorRowException(NAME, e.getMessage(), 1); } } } return results; }  @Override UsageDefinition define(); @Override void initialize(Arguments args); @Override void destroy(); @Override List<Row> execute(List<Row> rows, ExecutorContext context); @Override Mutation lineage(); static void jsonFlatten(JsonObject root, String field, int depth, int maxDepth, Row row); static Object getValue(JsonElement element); static Object getValue(JsonPrimitive primitive); }
JsParser implements Directive, Lineage { @Override public List<Row> execute(List<Row> rows, ExecutorContext context) throws DirectiveExecutionException, ErrorRowException { List<Row> results = new ArrayList<>(); for (Row row : rows) { int idx = row.find(column); if (idx != -1) { Object value = row.getValue(idx); if (value == null) { continue; } try { JsonElement element = null; if (value instanceof String) { String document = (String) value; element = parser.parse(document.trim()); } else if (value instanceof JsonObject || value instanceof JsonArray) { element = (JsonElement) value; } else { throw new DirectiveExecutionException( NAME, String.format("Column '%s' is of invalid type '%s'. It should be of type 'String'" + " or 'JsonObject' or 'JsonArray'.", column, value.getClass().getSimpleName())); } row.remove(idx); if (element != null) { if (element instanceof JsonObject) { jsonFlatten(element.getAsJsonObject(), column, 1, depth, row); results.add(row); } else if (element instanceof JsonArray) { JsonArray array = element.getAsJsonArray(); if (array.size() > 0) { for (int i = 0; i < array.size(); ++i) { JsonElement object = array.get(i); Row newRow = new Row(row); newRow.add(column, getValue(object)); results.add(newRow); } } else { results.add(row); } } else if (element instanceof JsonPrimitive) { row.add(column, getValue(element.getAsJsonPrimitive())); } } } catch (JSONException e) { throw new ErrorRowException(NAME, e.getMessage(), 1); } } } return results; }  @Override UsageDefinition define(); @Override void initialize(Arguments args); @Override void destroy(); @Override List<Row> execute(List<Row> rows, ExecutorContext context); @Override Mutation lineage(); static void jsonFlatten(JsonObject root, String field, int depth, int maxDepth, Row row); static Object getValue(JsonElement element); static Object getValue(JsonPrimitive primitive); static final String NAME; }