class 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);
		}
	}

	public  AnyToTripleFlatMapper(TableSchema dataSchema, Params params);

}