@Test public void testParseJsonAndJsonPath() throws Exception { String[] directives = new String[] { "parse-as-json body", "parse-as-csv body_deviceReference_screenSize | false", "drop body_deviceReference_screenSize", "rename body_deviceReference_screenSize_1 size1", "rename body_deviceReference_screenSize_2 size2", "rename body_deviceReference_screenSize_3 size3", "rename body_deviceReference_screenSize_4 size4", "json-path body_deviceReference_alerts signal_lost $.[*].['Signal lost']", "json-path signal_lost signal_lost $.[0]", "drop body", "rename body_deviceReference_timestamp timestamp", "set column timestamp timestamp", "drop body_deviceReference_alerts" }; List<Row> rows = Arrays.asList( new Row("body", "{ \"deviceReference\": { \"brand\": \"Samsung \", \"type\": \"Gear S3 frontier\", " + "\"deviceId\": \"SM-R760NDAAXAR\", \"timestamp\": 122121212341231, \"OS\": { \"name\": \"Tizen OS\", " + "\"version\": \"2.3.1\" }, \"alerts\": [ { \"Signal lost\": true }, { \"Emergency call\": true }, " + "{ \"Wifi connection lost\": true }, { \"Battery low\": true }, { \"Calories\": 354 } ], \"screenSize\": " + "\"extra-small|small|medium|large\", \"battery\": \"22%\", \"telephoneNumber\": \"+14099594986\", " + "\"comments\": \"It is an AT&T samung wearable device.\" } }") ); 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; }