AnyToTripleFlatMapper extends FlatMapper implements Serializable { @Override public void flatMap(Row row, Collector <Row> output) throws Exception { if (null == row) { output.collect(null); } bufMap.clear(); boolean success = formatReader.read(row, bufMap); if (success) { for (Map.Entry <String, String> entry : bufMap.entrySet()) { Tuple2 <Boolean, Object> parsedKey = ColumnsWriter.parseField(parsers[0], entry.getKey(), isString[0]); Tuple2 <Boolean, Object> parsedValue = ColumnsWriter.parseField(parsers[1], entry.getValue(), isString[1]); if (!StringUtils.isNullOrWhitespaceOnly(entry.getValue())) { if (parsedKey.f0 && parsedValue.f0) { output.collect(outputColsHelper .getResultRow(row, Row.of(parsedKey.f1, parsedValue.f1))); } else if (handleInvalid.equals(HandleInvalid.ERROR)) { throw new RuntimeException("Fail to write: " + JsonConverter.toJson(bufMap)); } } } } else if (handleInvalid.equals(HandleInvalid.ERROR)) { throw new RuntimeException("Fail to read: " + row); } } AnyToTripleFlatMapper(TableSchema dataSchema, Params params); @Override void open(); @Override void flatMap(Row row, Collector <Row> output); @Override TableSchema getOutputSchema();  }
@Test public void flatMap() throws Exception { AnyToTripleFlatMapper transKvToTriple = new AnyToTripleFlatMapper( CsvUtil.schemaStr2Schema("row_id long, kv string"), new Params() .set(FormatTransParams.FROM_FORMAT, FormatType.KV) .set(FromKvParams.KV_COL, "kv") .set(FromKvParams.KV_COL_DELIMITER, ",") .set(FromKvParams.KV_VAL_DELIMITER, ":") .set(ToTripleParams.TRIPLE_COL_VAL_SCHEMA_STR, "col_id int, val double") .set(ToTripleParams.RESERVED_COLS, new String[] {"row_id"}) ); transKvToTriple.open(); RowCollector collector = new RowCollector(); transKvToTriple.flatMap(Row.of(3L, "1:1.0,4:1.0"), collector); Assert.assertEquals(collector.getRows().size(), 2); }