@Test public void testEmbeddedValueFaultTolerant() throws Exception { userNeedColNames = new String[] { "user_id", "nonexisted_description" }; userNeedColNamesComment = new String[] { "", "" }; List<TblColRef> allCol = mockupTblColRefList(); TimedJsonStreamParser parser = new TimedJsonStreamParser(allCol, null); Object msg = mapper.readValue(new File(jsonFilePath), mapType); ByteBuffer buffer = getJsonByteBuffer(msg); List<StreamingMessageRow> msgList = parser.parse(buffer); List<String> result = msgList.get(0).getData(); assertEquals("4853763947", result.get(0)); assertEquals(StringUtils.EMPTY, result.get(1)); }
@Override public List<StreamingMessageRow> parse(ByteBuffer buffer) { try { Map<String, Object> message = mapper.readValue(new ByteBufferBackedInputStream(buffer), mapType); root.clear(); root.putAll(message); String tsStr = objToString(root.get(tsColName)); long t = streamTimeParser.parseTime(tsStr); ArrayList<String> result = Lists.newArrayList(); for (TblColRef column : allColumns) { final String columnName = colLowerCaseMap.get(column.getName()); if (populateDerivedTimeColumns(columnName, result, t) == false) { result.add(getValueByKey(column, root)); } } StreamingMessageRow streamingMessageRow = new StreamingMessageRow(result, 0, t, Collections.<String, Object>emptyMap()); List<StreamingMessageRow> messageRowList = new ArrayList<StreamingMessageRow>(); messageRowList.add(streamingMessageRow); return messageRowList; } catch (IOException e) { logger.error("malformed data: {}", new String(buffer.array(), Charsets.UTF_8)); logger.error("error", e); throw new RuntimeException(e); } }
TimedJsonStreamParser extends StreamingParser { @Override public List<StreamingMessageRow> parse(ByteBuffer buffer) { try { Map<String, Object> message = mapper.readValue(new ByteBufferBackedInputStream(buffer), mapType); root.clear(); root.putAll(message); String tsStr = objToString(root.get(tsColName)); long t = streamTimeParser.parseTime(tsStr); ArrayList<String> result = Lists.newArrayList(); for (TblColRef column : allColumns) { final String columnName = colLowerCaseMap.get(column.getName()); if (populateDerivedTimeColumns(columnName, result, t) == false) { result.add(getValueByKey(column, root)); } } StreamingMessageRow streamingMessageRow = new StreamingMessageRow(result, 0, t, Collections.<String, Object>emptyMap()); List<StreamingMessageRow> messageRowList = new ArrayList<StreamingMessageRow>(); messageRowList.add(streamingMessageRow); return messageRowList; } catch (IOException e) { logger.error("malformed data: {}", new String(buffer.array(), Charsets.UTF_8)); logger.error("error", e); throw new RuntimeException(e); } } }
TimedJsonStreamParser extends StreamingParser { @Override public List<StreamingMessageRow> parse(ByteBuffer buffer) { try { Map<String, Object> message = mapper.readValue(new ByteBufferBackedInputStream(buffer), mapType); root.clear(); root.putAll(message); String tsStr = objToString(root.get(tsColName)); long t = streamTimeParser.parseTime(tsStr); ArrayList<String> result = Lists.newArrayList(); for (TblColRef column : allColumns) { final String columnName = colLowerCaseMap.get(column.getName()); if (populateDerivedTimeColumns(columnName, result, t) == false) { result.add(getValueByKey(column, root)); } } StreamingMessageRow streamingMessageRow = new StreamingMessageRow(result, 0, t, Collections.<String, Object>emptyMap()); List<StreamingMessageRow> messageRowList = new ArrayList<StreamingMessageRow>(); messageRowList.add(streamingMessageRow); return messageRowList; } catch (IOException e) { logger.error("malformed data: {}", new String(buffer.array(), Charsets.UTF_8)); logger.error("error", e); throw new RuntimeException(e); } } TimedJsonStreamParser(List<TblColRef> allColumns, Map<String, String> properties); }
TimedJsonStreamParser extends StreamingParser { @Override public List<StreamingMessageRow> parse(ByteBuffer buffer) { try { Map<String, Object> message = mapper.readValue(new ByteBufferBackedInputStream(buffer), mapType); root.clear(); root.putAll(message); String tsStr = objToString(root.get(tsColName)); long t = streamTimeParser.parseTime(tsStr); ArrayList<String> result = Lists.newArrayList(); for (TblColRef column : allColumns) { final String columnName = colLowerCaseMap.get(column.getName()); if (populateDerivedTimeColumns(columnName, result, t) == false) { result.add(getValueByKey(column, root)); } } StreamingMessageRow streamingMessageRow = new StreamingMessageRow(result, 0, t, Collections.<String, Object>emptyMap()); List<StreamingMessageRow> messageRowList = new ArrayList<StreamingMessageRow>(); messageRowList.add(streamingMessageRow); return messageRowList; } catch (IOException e) { logger.error("malformed data: {}", new String(buffer.array(), Charsets.UTF_8)); logger.error("error", e); throw new RuntimeException(e); } } TimedJsonStreamParser(List<TblColRef> allColumns, Map<String, String> properties); @Override List<StreamingMessageRow> parse(ByteBuffer buffer); @Override boolean filter(StreamingMessageRow streamingMessageRow); String[] getEmbeddedPropertyNames(TblColRef column); static String objToString(Object value); }
TimedJsonStreamParser extends StreamingParser { @Override public List<StreamingMessageRow> parse(ByteBuffer buffer) { try { Map<String, Object> message = mapper.readValue(new ByteBufferBackedInputStream(buffer), mapType); root.clear(); root.putAll(message); String tsStr = objToString(root.get(tsColName)); long t = streamTimeParser.parseTime(tsStr); ArrayList<String> result = Lists.newArrayList(); for (TblColRef column : allColumns) { final String columnName = colLowerCaseMap.get(column.getName()); if (populateDerivedTimeColumns(columnName, result, t) == false) { result.add(getValueByKey(column, root)); } } StreamingMessageRow streamingMessageRow = new StreamingMessageRow(result, 0, t, Collections.<String, Object>emptyMap()); List<StreamingMessageRow> messageRowList = new ArrayList<StreamingMessageRow>(); messageRowList.add(streamingMessageRow); return messageRowList; } catch (IOException e) { logger.error("malformed data: {}", new String(buffer.array(), Charsets.UTF_8)); logger.error("error", e); throw new RuntimeException(e); } } TimedJsonStreamParser(List<TblColRef> allColumns, Map<String, String> properties); @Override List<StreamingMessageRow> parse(ByteBuffer buffer); @Override boolean filter(StreamingMessageRow streamingMessageRow); String[] getEmbeddedPropertyNames(TblColRef column); static String objToString(Object value); static final String EMBEDDED_PROPERTY_SEPARATOR; }