@Test public void testSerialize() throws StreamSerDeException { CSVSerDe csvSerDe = new CSVSerDe(); csvSerDe.setSchema(schema); SimpleSerDe deser = new SimpleSerDe(); deser.setSchema(schema); String s = "3,\"a,b\",\"\"\"a,b\"\"\""; List<Object[]> results= csvSerDe.deSerialize(s); assertTrue(csvSerDe.serialize(results).equals(s)); }
@Override public Object serialize(List< Object[] > events) throws StreamSerDeException { if (events == null) { LOG.info("Input event is null."); return null; } sb.delete(0, sb.length()); for (Object[] event : events) { String value = csvCreator.createCSV(event); if (value != null) { sb.append(value + LINE_SEPARATOR_UNIX); } } return sb.substring(0, sb.length() - LINE_SEPARATOR_UNIX.length()); }
CSVSerDe extends BaseSerDe { @Override public Object serialize(List< Object[] > events) throws StreamSerDeException { if (events == null) { LOG.info("Input event is null."); return null; } sb.delete(0, sb.length()); for (Object[] event : events) { String value = csvCreator.createCSV(event); if (value != null) { sb.append(value + LINE_SEPARATOR_UNIX); } } return sb.substring(0, sb.length() - LINE_SEPARATOR_UNIX.length()); } }
CSVSerDe extends BaseSerDe { @Override public Object serialize(List< Object[] > events) throws StreamSerDeException { if (events == null) { LOG.info("Input event is null."); return null; } sb.delete(0, sb.length()); for (Object[] event : events) { String value = csvCreator.createCSV(event); if (value != null) { sb.append(value + LINE_SEPARATOR_UNIX); } } return sb.substring(0, sb.length() - LINE_SEPARATOR_UNIX.length()); }  }
CSVSerDe extends BaseSerDe { @Override public Object serialize(List< Object[] > events) throws StreamSerDeException { if (events == null) { LOG.info("Input event is null."); return null; } sb.delete(0, sb.length()); for (Object[] event : events) { String value = csvCreator.createCSV(event); if (value != null) { sb.append(value + LINE_SEPARATOR_UNIX); } } return sb.substring(0, sb.length() - LINE_SEPARATOR_UNIX.length()); }  @Override void setConfig(StreamingConfig conf); @Override void initialize(); @Override List< Object[] > deSerialize(Object data); @Override Object serialize(List< Object[] > events); }
CSVSerDe extends BaseSerDe { @Override public Object serialize(List< Object[] > events) throws StreamSerDeException { if (events == null) { LOG.info("Input event is null."); return null; } sb.delete(0, sb.length()); for (Object[] event : events) { String value = csvCreator.createCSV(event); if (value != null) { sb.append(value + LINE_SEPARATOR_UNIX); } } return sb.substring(0, sb.length() - LINE_SEPARATOR_UNIX.length()); }  @Override void setConfig(StreamingConfig conf); @Override void initialize(); @Override List< Object[] > deSerialize(Object data); @Override Object serialize(List< Object[] > events); static final String LINE_SEPARATOR_UNIX; }