@Test public void testSchema() throws IOException { final TypeInformation<Row> rowSchema = JsonRowSchemaConverter.convert( "{" + " type: 'object'," + " properties: {" + " id: { type: 'integer' }," + " idNumber: { type: 'number' }," + " idOrNull: { type: ['integer', 'null'] }," + " name: { type: 'string' }," + " date: { type: 'string', format: 'date' }," + " time: { type: 'string', format: 'time' }," + " timestamp: { type: 'string', format: 'date-time' }," + " bytes: { type: 'string', contentEncoding: 'base64' }," + " numbers: { type: 'array', items: { type: 'integer' } }," + " strings: { type: 'array', items: { type: 'string' } }," + " nested: { " + " type: 'object'," + " properties: { " + " booleanField: { type: 'boolean' }," + " decimalField: { type: 'number' }" + " }" + " }" + " }" + "}"); final Row row = new Row(11); row.setField(0, BigDecimal.valueOf(-333)); row.setField(1, BigDecimal.valueOf(12.2222)); row.setField(2, null); row.setField(3, ""); row.setField(4, Date.valueOf("1990-10-14")); row.setField(5, Time.valueOf("12:12:43")); row.setField(6, Timestamp.valueOf("1990-10-14 12:12:43")); final byte[] bytes = new byte[1024]; ThreadLocalRandom.current().nextBytes(bytes); row.setField(7, bytes); final BigDecimal[] numbers = new BigDecimal[] { BigDecimal.valueOf(1), BigDecimal.valueOf(2), BigDecimal.valueOf(3)}; row.setField(8, numbers); final String[] strings = new String[] {"one", "two", "three"}; row.setField(9, strings); final Row nestedRow = new Row(2); nestedRow.setField(0, true); nestedRow.setField(1, BigDecimal.valueOf(12)); row.setField(10, nestedRow); final Row resultRow = serializeAndDeserialize(rowSchema, row); assertEquals(row, resultRow); }
private JsonNode convert(ContainerNode<?> container, JsonNode reuse, TypeInformation<?> info, Object object) { if (info == Types.VOID || object == null) { return container.nullNode(); } else if (info == Types.BOOLEAN) { return container.booleanNode((Boolean) object); } else if (info == Types.STRING) { return container.textNode((String) object); } else if (info == Types.BIG_DEC) { if (object instanceof BigDecimal) { return container.numberNode((BigDecimal) object); } return container.numberNode(BigDecimal.valueOf(((Number) object).doubleValue())); } else if (info == Types.BIG_INT) { if (object instanceof BigInteger) { return container.numberNode((BigInteger) object); } return container.numberNode(BigInteger.valueOf(((Number) object).longValue())); } else if (info == Types.SQL_DATE) { return container.textNode(object.toString()); } else if (info == Types.SQL_TIME) { final Time time = (Time) object; if (time.getTime() % 1000 > 0) { return container.textNode(timeFormatWithMillis.format(time)); } return container.textNode(timeFormat.format(time)); } else if (info == Types.SQL_TIMESTAMP) { return container.textNode(timestampFormat.format((Timestamp) object)); } else if (info instanceof RowTypeInfo) { if (reuse != null && reuse instanceof ObjectNode) { return convertRow((ObjectNode) reuse, (RowTypeInfo) info, (Row) object); } else { return convertRow(null, (RowTypeInfo) info, (Row) object); } } else if (info instanceof ObjectArrayTypeInfo) { if (reuse != null && reuse instanceof ArrayNode) { return convertObjectArray((ArrayNode) reuse, ((ObjectArrayTypeInfo) info).getComponentInfo(), (Object[]) object); } else { return convertObjectArray(null, ((ObjectArrayTypeInfo) info).getComponentInfo(), (Object[]) object); } } else if (info instanceof BasicArrayTypeInfo) { if (reuse != null && reuse instanceof ArrayNode) { return convertObjectArray((ArrayNode) reuse, ((BasicArrayTypeInfo) info).getComponentInfo(), (Object[]) object); } else { return convertObjectArray(null, ((BasicArrayTypeInfo) info).getComponentInfo(), (Object[]) object); } } else if (info instanceof PrimitiveArrayTypeInfo && ((PrimitiveArrayTypeInfo) info).getComponentType() == Types.BYTE) { return container.binaryNode((byte[]) object); } else { try { return mapper.valueToTree(object); } catch (IllegalArgumentException e) { throw new IllegalStateException("Unsupported type information '" + info + "' for object: " + object, e); } } }
JsonRowSerializationSchema implements SerializationSchema<Row> { private JsonNode convert(ContainerNode<?> container, JsonNode reuse, TypeInformation<?> info, Object object) { if (info == Types.VOID || object == null) { return container.nullNode(); } else if (info == Types.BOOLEAN) { return container.booleanNode((Boolean) object); } else if (info == Types.STRING) { return container.textNode((String) object); } else if (info == Types.BIG_DEC) { if (object instanceof BigDecimal) { return container.numberNode((BigDecimal) object); } return container.numberNode(BigDecimal.valueOf(((Number) object).doubleValue())); } else if (info == Types.BIG_INT) { if (object instanceof BigInteger) { return container.numberNode((BigInteger) object); } return container.numberNode(BigInteger.valueOf(((Number) object).longValue())); } else if (info == Types.SQL_DATE) { return container.textNode(object.toString()); } else if (info == Types.SQL_TIME) { final Time time = (Time) object; if (time.getTime() % 1000 > 0) { return container.textNode(timeFormatWithMillis.format(time)); } return container.textNode(timeFormat.format(time)); } else if (info == Types.SQL_TIMESTAMP) { return container.textNode(timestampFormat.format((Timestamp) object)); } else if (info instanceof RowTypeInfo) { if (reuse != null && reuse instanceof ObjectNode) { return convertRow((ObjectNode) reuse, (RowTypeInfo) info, (Row) object); } else { return convertRow(null, (RowTypeInfo) info, (Row) object); } } else if (info instanceof ObjectArrayTypeInfo) { if (reuse != null && reuse instanceof ArrayNode) { return convertObjectArray((ArrayNode) reuse, ((ObjectArrayTypeInfo) info).getComponentInfo(), (Object[]) object); } else { return convertObjectArray(null, ((ObjectArrayTypeInfo) info).getComponentInfo(), (Object[]) object); } } else if (info instanceof BasicArrayTypeInfo) { if (reuse != null && reuse instanceof ArrayNode) { return convertObjectArray((ArrayNode) reuse, ((BasicArrayTypeInfo) info).getComponentInfo(), (Object[]) object); } else { return convertObjectArray(null, ((BasicArrayTypeInfo) info).getComponentInfo(), (Object[]) object); } } else if (info instanceof PrimitiveArrayTypeInfo && ((PrimitiveArrayTypeInfo) info).getComponentType() == Types.BYTE) { return container.binaryNode((byte[]) object); } else { try { return mapper.valueToTree(object); } catch (IllegalArgumentException e) { throw new IllegalStateException("Unsupported type information '" + info + "' for object: " + object, e); } } } }
JsonRowSerializationSchema implements SerializationSchema<Row> { private JsonNode convert(ContainerNode<?> container, JsonNode reuse, TypeInformation<?> info, Object object) { if (info == Types.VOID || object == null) { return container.nullNode(); } else if (info == Types.BOOLEAN) { return container.booleanNode((Boolean) object); } else if (info == Types.STRING) { return container.textNode((String) object); } else if (info == Types.BIG_DEC) { if (object instanceof BigDecimal) { return container.numberNode((BigDecimal) object); } return container.numberNode(BigDecimal.valueOf(((Number) object).doubleValue())); } else if (info == Types.BIG_INT) { if (object instanceof BigInteger) { return container.numberNode((BigInteger) object); } return container.numberNode(BigInteger.valueOf(((Number) object).longValue())); } else if (info == Types.SQL_DATE) { return container.textNode(object.toString()); } else if (info == Types.SQL_TIME) { final Time time = (Time) object; if (time.getTime() % 1000 > 0) { return container.textNode(timeFormatWithMillis.format(time)); } return container.textNode(timeFormat.format(time)); } else if (info == Types.SQL_TIMESTAMP) { return container.textNode(timestampFormat.format((Timestamp) object)); } else if (info instanceof RowTypeInfo) { if (reuse != null && reuse instanceof ObjectNode) { return convertRow((ObjectNode) reuse, (RowTypeInfo) info, (Row) object); } else { return convertRow(null, (RowTypeInfo) info, (Row) object); } } else if (info instanceof ObjectArrayTypeInfo) { if (reuse != null && reuse instanceof ArrayNode) { return convertObjectArray((ArrayNode) reuse, ((ObjectArrayTypeInfo) info).getComponentInfo(), (Object[]) object); } else { return convertObjectArray(null, ((ObjectArrayTypeInfo) info).getComponentInfo(), (Object[]) object); } } else if (info instanceof BasicArrayTypeInfo) { if (reuse != null && reuse instanceof ArrayNode) { return convertObjectArray((ArrayNode) reuse, ((BasicArrayTypeInfo) info).getComponentInfo(), (Object[]) object); } else { return convertObjectArray(null, ((BasicArrayTypeInfo) info).getComponentInfo(), (Object[]) object); } } else if (info instanceof PrimitiveArrayTypeInfo && ((PrimitiveArrayTypeInfo) info).getComponentType() == Types.BYTE) { return container.binaryNode((byte[]) object); } else { try { return mapper.valueToTree(object); } catch (IllegalArgumentException e) { throw new IllegalStateException("Unsupported type information '" + info + "' for object: " + object, e); } } } JsonRowSerializationSchema(TypeInformation<Row> typeInfo);  JsonRowSerializationSchema(String jsonSchema); }
JsonRowSerializationSchema implements SerializationSchema<Row> { private JsonNode convert(ContainerNode<?> container, JsonNode reuse, TypeInformation<?> info, Object object) { if (info == Types.VOID || object == null) { return container.nullNode(); } else if (info == Types.BOOLEAN) { return container.booleanNode((Boolean) object); } else if (info == Types.STRING) { return container.textNode((String) object); } else if (info == Types.BIG_DEC) { if (object instanceof BigDecimal) { return container.numberNode((BigDecimal) object); } return container.numberNode(BigDecimal.valueOf(((Number) object).doubleValue())); } else if (info == Types.BIG_INT) { if (object instanceof BigInteger) { return container.numberNode((BigInteger) object); } return container.numberNode(BigInteger.valueOf(((Number) object).longValue())); } else if (info == Types.SQL_DATE) { return container.textNode(object.toString()); } else if (info == Types.SQL_TIME) { final Time time = (Time) object; if (time.getTime() % 1000 > 0) { return container.textNode(timeFormatWithMillis.format(time)); } return container.textNode(timeFormat.format(time)); } else if (info == Types.SQL_TIMESTAMP) { return container.textNode(timestampFormat.format((Timestamp) object)); } else if (info instanceof RowTypeInfo) { if (reuse != null && reuse instanceof ObjectNode) { return convertRow((ObjectNode) reuse, (RowTypeInfo) info, (Row) object); } else { return convertRow(null, (RowTypeInfo) info, (Row) object); } } else if (info instanceof ObjectArrayTypeInfo) { if (reuse != null && reuse instanceof ArrayNode) { return convertObjectArray((ArrayNode) reuse, ((ObjectArrayTypeInfo) info).getComponentInfo(), (Object[]) object); } else { return convertObjectArray(null, ((ObjectArrayTypeInfo) info).getComponentInfo(), (Object[]) object); } } else if (info instanceof BasicArrayTypeInfo) { if (reuse != null && reuse instanceof ArrayNode) { return convertObjectArray((ArrayNode) reuse, ((BasicArrayTypeInfo) info).getComponentInfo(), (Object[]) object); } else { return convertObjectArray(null, ((BasicArrayTypeInfo) info).getComponentInfo(), (Object[]) object); } } else if (info instanceof PrimitiveArrayTypeInfo && ((PrimitiveArrayTypeInfo) info).getComponentType() == Types.BYTE) { return container.binaryNode((byte[]) object); } else { try { return mapper.valueToTree(object); } catch (IllegalArgumentException e) { throw new IllegalStateException("Unsupported type information '" + info + "' for object: " + object, e); } } } JsonRowSerializationSchema(TypeInformation<Row> typeInfo);  JsonRowSerializationSchema(String jsonSchema); @Override byte[] serialize(Row row); @Override boolean equals(Object o); @Override int hashCode(); }
JsonRowSerializationSchema implements SerializationSchema<Row> { private JsonNode convert(ContainerNode<?> container, JsonNode reuse, TypeInformation<?> info, Object object) { if (info == Types.VOID || object == null) { return container.nullNode(); } else if (info == Types.BOOLEAN) { return container.booleanNode((Boolean) object); } else if (info == Types.STRING) { return container.textNode((String) object); } else if (info == Types.BIG_DEC) { if (object instanceof BigDecimal) { return container.numberNode((BigDecimal) object); } return container.numberNode(BigDecimal.valueOf(((Number) object).doubleValue())); } else if (info == Types.BIG_INT) { if (object instanceof BigInteger) { return container.numberNode((BigInteger) object); } return container.numberNode(BigInteger.valueOf(((Number) object).longValue())); } else if (info == Types.SQL_DATE) { return container.textNode(object.toString()); } else if (info == Types.SQL_TIME) { final Time time = (Time) object; if (time.getTime() % 1000 > 0) { return container.textNode(timeFormatWithMillis.format(time)); } return container.textNode(timeFormat.format(time)); } else if (info == Types.SQL_TIMESTAMP) { return container.textNode(timestampFormat.format((Timestamp) object)); } else if (info instanceof RowTypeInfo) { if (reuse != null && reuse instanceof ObjectNode) { return convertRow((ObjectNode) reuse, (RowTypeInfo) info, (Row) object); } else { return convertRow(null, (RowTypeInfo) info, (Row) object); } } else if (info instanceof ObjectArrayTypeInfo) { if (reuse != null && reuse instanceof ArrayNode) { return convertObjectArray((ArrayNode) reuse, ((ObjectArrayTypeInfo) info).getComponentInfo(), (Object[]) object); } else { return convertObjectArray(null, ((ObjectArrayTypeInfo) info).getComponentInfo(), (Object[]) object); } } else if (info instanceof BasicArrayTypeInfo) { if (reuse != null && reuse instanceof ArrayNode) { return convertObjectArray((ArrayNode) reuse, ((BasicArrayTypeInfo) info).getComponentInfo(), (Object[]) object); } else { return convertObjectArray(null, ((BasicArrayTypeInfo) info).getComponentInfo(), (Object[]) object); } } else if (info instanceof PrimitiveArrayTypeInfo && ((PrimitiveArrayTypeInfo) info).getComponentType() == Types.BYTE) { return container.binaryNode((byte[]) object); } else { try { return mapper.valueToTree(object); } catch (IllegalArgumentException e) { throw new IllegalStateException("Unsupported type information '" + info + "' for object: " + object, e); } } } JsonRowSerializationSchema(TypeInformation<Row> typeInfo);  JsonRowSerializationSchema(String jsonSchema); @Override byte[] serialize(Row row); @Override boolean equals(Object o); @Override int hashCode();  }