You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2018/02/15 19:18:15 UTC
[1/2] flink git commit: [FLINK-8630] [formats] Add proper support for
JSON formats
Repository: flink
Updated Branches:
refs/heads/master 942649e93 -> 6c09c0578
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowDeserializationSchemaTest.java
----------------------------------------------------------------------
diff --git a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowDeserializationSchemaTest.java b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowDeserializationSchemaTest.java
new file mode 100644
index 0000000..1c4a17a
--- /dev/null
+++ b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowDeserializationSchemaTest.java
@@ -0,0 +1,198 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.formats.json;
+
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.types.Row;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.sql.Date;
+import java.sql.Time;
+import java.sql.Timestamp;
+import java.util.concurrent.ThreadLocalRandom;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Tests for the {@link JsonRowDeserializationSchema}.
+ */
+public class JsonRowDeserializationSchemaTest {
+
+ /**
+ * Tests simple deserialization using type information.
+ */
+ @Test
+ public void testTypeInfoDeserialization() throws Exception {
+ long id = 1238123899121L;
+ String name = "asdlkjasjkdla998y1122";
+ byte[] bytes = new byte[1024];
+ ThreadLocalRandom.current().nextBytes(bytes);
+
+ ObjectMapper objectMapper = new ObjectMapper();
+
+ // Root
+ ObjectNode root = objectMapper.createObjectNode();
+ root.put("id", id);
+ root.put("name", name);
+ root.put("bytes", bytes);
+
+ byte[] serializedJson = objectMapper.writeValueAsBytes(root);
+
+ JsonRowDeserializationSchema deserializationSchema = new JsonRowDeserializationSchema(
+ Types.ROW_NAMED(
+ new String[] { "id", "name", "bytes" },
+ Types.LONG, Types.STRING, Types.PRIMITIVE_ARRAY(Types.BYTE))
+ );
+
+ Row deserialized = deserializationSchema.deserialize(serializedJson);
+
+ assertEquals(3, deserialized.getArity());
+ assertEquals(id, deserialized.getField(0));
+ assertEquals(name, deserialized.getField(1));
+ assertArrayEquals(bytes, (byte[]) deserialized.getField(2));
+ }
+
+ @Test
+ public void testSchemaDeserialization() throws Exception {
+ final BigInteger id = BigInteger.valueOf(1238123899121L);
+ final String name = "asdlkjasjkdla998y1122";
+ final byte[] bytes = new byte[1024];
+ ThreadLocalRandom.current().nextBytes(bytes);
+ final BigInteger[] numbers = new BigInteger[] {
+ BigInteger.valueOf(1), BigInteger.valueOf(2), BigInteger.valueOf(3)};
+ final String[] strings = new String[] {"one", "two", "three"};
+
+ final ObjectMapper objectMapper = new ObjectMapper();
+
+ // Root
+ ObjectNode root = objectMapper.createObjectNode();
+ root.put("id", id.longValue());
+ root.putNull("idOrNull");
+ root.put("name", name);
+ root.put("date", "1990-10-14");
+ root.put("time", "12:12:43Z");
+ root.put("timestamp", "1990-10-14T12:12:43Z");
+ root.put("bytes", bytes);
+ root.putArray("numbers").add(1).add(2).add(3);
+ root.putArray("strings").add("one").add("two").add("three");
+ root.putObject("nested").put("booleanField", true).put("decimalField", 12);
+
+ final byte[] serializedJson = objectMapper.writeValueAsBytes(root);
+
+ JsonRowDeserializationSchema deserializationSchema = new JsonRowDeserializationSchema(
+ "{" +
+ " type: 'object'," +
+ " properties: {" +
+ " id: { type: 'integer' }," +
+ " 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 deserialized = deserializationSchema.deserialize(serializedJson);
+
+ final Row expected = new Row(10);
+ expected.setField(0, id);
+ expected.setField(1, null);
+ expected.setField(2, name);
+ expected.setField(3, Date.valueOf("1990-10-14"));
+ expected.setField(4, Time.valueOf("12:12:43"));
+ expected.setField(5, Timestamp.valueOf("1990-10-14 12:12:43"));
+ expected.setField(6, bytes);
+ expected.setField(7, numbers);
+ expected.setField(8, strings);
+ final Row nestedRow = new Row(2);
+ nestedRow.setField(0, true);
+ nestedRow.setField(1, BigDecimal.valueOf(12));
+ expected.setField(9, nestedRow);
+
+ assertEquals(expected, deserialized);
+ }
+
+ /**
+ * Tests deserialization with non-existing field name.
+ */
+ @Test
+ public void testMissingNode() throws Exception {
+ ObjectMapper objectMapper = new ObjectMapper();
+
+ // Root
+ ObjectNode root = objectMapper.createObjectNode();
+ root.put("id", 123123123);
+ byte[] serializedJson = objectMapper.writeValueAsBytes(root);
+
+ JsonRowDeserializationSchema deserializationSchema = new JsonRowDeserializationSchema(
+ Types.ROW_NAMED(
+ new String[] { "name" },
+ Types.STRING)
+ );
+
+ Row row = deserializationSchema.deserialize(serializedJson);
+
+ assertEquals(1, row.getArity());
+ Assert.assertNull("Missing field not null", row.getField(0));
+
+ deserializationSchema.setFailOnMissingField(true);
+
+ try {
+ deserializationSchema.deserialize(serializedJson);
+ Assert.fail("Did not throw expected Exception");
+ } catch (IOException e) {
+ Assert.assertTrue(e.getCause() instanceof IllegalStateException);
+ }
+ }
+
+ /**
+ * Tests that number of field names and types has to match.
+ */
+ @Test
+ public void testNumberOfFieldNamesAndTypesMismatch() {
+ try {
+ new JsonRowDeserializationSchema(
+ Types.ROW_NAMED(
+ new String[]{"one", "two", "three"},
+ Types.LONG));
+ Assert.fail("Did not throw expected Exception");
+ } catch (IllegalArgumentException ignored) {
+ // Expected
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowSerializationSchemaTest.java
----------------------------------------------------------------------
diff --git a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowSerializationSchemaTest.java b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowSerializationSchemaTest.java
new file mode 100644
index 0000000..f956fc6
--- /dev/null
+++ b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowSerializationSchemaTest.java
@@ -0,0 +1,174 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.formats.json;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.types.Row;
+
+import org.junit.Test;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.sql.Date;
+import java.sql.Time;
+import java.sql.Timestamp;
+import java.util.concurrent.ThreadLocalRandom;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Tests for the {@link JsonRowSerializationSchema}.
+ */
+public class JsonRowSerializationSchemaTest {
+
+ @Test
+ public void testRowSerialization() throws IOException {
+ final TypeInformation<Row> rowSchema = Types.ROW_NAMED(
+ new String[] {"f1", "f2", "f3"},
+ Types.INT, Types.BOOLEAN, Types.STRING);
+
+ final Row row = new Row(3);
+ row.setField(0, 1);
+ row.setField(1, true);
+ row.setField(2, "str");
+
+ final Row resultRow = serializeAndDeserialize(rowSchema, row);
+ assertEquals(row, resultRow);
+ }
+
+ @Test
+ public void testSerializationOfTwoRows() throws IOException {
+ final TypeInformation<Row> rowSchema = Types.ROW_NAMED(
+ new String[] {"f1", "f2", "f3"},
+ Types.INT, Types.BOOLEAN, Types.STRING);
+
+ final Row row1 = new Row(3);
+ row1.setField(0, 1);
+ row1.setField(1, true);
+ row1.setField(2, "str");
+
+ final JsonRowSerializationSchema serializationSchema = new JsonRowSerializationSchema(rowSchema);
+ final JsonRowDeserializationSchema deserializationSchema = new JsonRowDeserializationSchema(rowSchema);
+
+ byte[] bytes = serializationSchema.serialize(row1);
+ assertEquals(row1, deserializationSchema.deserialize(bytes));
+
+ final Row row2 = new Row(3);
+ row2.setField(0, 10);
+ row2.setField(1, false);
+ row2.setField(2, "newStr");
+
+ bytes = serializationSchema.serialize(row2);
+ assertEquals(row2, deserializationSchema.deserialize(bytes));
+ }
+
+ @Test
+ public void testNestedSchema() throws IOException {
+ final TypeInformation<Row> rowSchema = Types.ROW_NAMED(
+ new String[] {"f1", "f2", "f3"},
+ Types.INT, Types.BOOLEAN, Types.ROW(Types.INT, Types.DOUBLE));
+
+ final Row row = new Row(3);
+ row.setField(0, 42);
+ row.setField(1, false);
+ final Row nested = new Row(2);
+ nested.setField(0, 22);
+ nested.setField(1, 2.3);
+ row.setField(2, nested);
+
+ final Row resultRow = serializeAndDeserialize(rowSchema, row);
+ assertEquals(row, resultRow);
+ }
+
+ @Test(expected = RuntimeException.class)
+ public void testSerializeRowWithInvalidNumberOfFields() {
+ final TypeInformation<Row> rowSchema = Types.ROW_NAMED(
+ new String[] {"f1", "f2", "f3"},
+ Types.INT, Types.BOOLEAN, Types.STRING);
+
+ final Row row = new Row(1);
+ row.setField(0, 1);
+
+ final JsonRowSerializationSchema serializationSchema = new JsonRowSerializationSchema(rowSchema);
+ serializationSchema.serialize(row);
+ }
+
+ @Test
+ public void testSchema() throws IOException {
+ final TypeInformation<Row> rowSchema = JsonSchemaConverter.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, BigInteger.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 BigInteger[] numbers = new BigInteger[] {
+ BigInteger.valueOf(1), BigInteger.valueOf(2), BigInteger.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 Row serializeAndDeserialize(TypeInformation<Row> rowSchema, Row row) throws IOException {
+ final JsonRowSerializationSchema serializationSchema = new JsonRowSerializationSchema(rowSchema);
+ final JsonRowDeserializationSchema deserializationSchema = new JsonRowDeserializationSchema(rowSchema);
+
+ final byte[] bytes = serializationSchema.serialize(row);
+ return deserializationSchema.deserialize(bytes);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonSchemaConverterTest.java
----------------------------------------------------------------------
diff --git a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonSchemaConverterTest.java b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonSchemaConverterTest.java
new file mode 100644
index 0000000..d072663
--- /dev/null
+++ b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonSchemaConverterTest.java
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.formats.json;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.util.FileUtils;
+
+import org.junit.Test;
+
+import java.io.File;
+import java.net.URL;
+import java.util.Objects;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Tests for {@link JsonSchemaConverter}.
+ */
+public class JsonSchemaConverterTest {
+
+ @Test
+ public void testComplexSchema() throws Exception {
+ final URL url = getClass().getClassLoader().getResource("complex-schema.json");
+ Objects.requireNonNull(url);
+ final String schema = FileUtils.readFileUtf8(new File(url.getFile()));
+ final TypeInformation<?> result = JsonSchemaConverter.convert(schema);
+
+ final TypeInformation<?> expected = Types.ROW_NAMED(
+ new String[] {"fn", "familyName", "additionalName", "tuples", "honorificPrefix", "url",
+ "email", "tel", "sound", "org"},
+ Types.STRING, Types.STRING, Types.BOOLEAN, Types.ROW(Types.BIG_DEC, Types.STRING, Types.STRING, Types.STRING),
+ Types.OBJECT_ARRAY(Types.STRING), Types.STRING, Types.ROW_NAMED(new String[] {"type", "value"}, Types.STRING, Types.STRING),
+ Types.ROW_NAMED(new String[] {"type", "value"}, Types.BIG_INT, Types.STRING), Types.VOID,
+ Types.ROW_NAMED(new String[] {"organizationUnit"}, Types.ROW()));
+
+ assertEquals(expected, result);
+ }
+
+ @Test
+ public void testReferenceSchema() throws Exception {
+ final URL url = getClass().getClassLoader().getResource("reference-schema.json");
+ Objects.requireNonNull(url);
+ final String schema = FileUtils.readFileUtf8(new File(url.getFile()));
+ final TypeInformation<?> result = JsonSchemaConverter.convert(schema);
+
+ final TypeInformation<?> expected = Types.ROW_NAMED(
+ new String[] {"billing_address", "shipping_address", "optional_address"},
+ Types.ROW_NAMED(new String[] {"street_address", "city", "state"}, Types.STRING, Types.STRING, Types.STRING),
+ Types.ROW_NAMED(new String[] {"street_address", "city", "state"}, Types.STRING, Types.STRING, Types.STRING),
+ Types.ROW_NAMED(new String[] {"street_address", "city", "state"}, Types.STRING, Types.STRING, Types.STRING));
+
+ assertEquals(expected, result);
+ }
+
+ @Test
+ public void testAtomicType() {
+ final TypeInformation<?> result = JsonSchemaConverter.convert("{ type: 'number' }");
+
+ assertEquals(Types.BIG_DEC, result);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testMissingType() {
+ JsonSchemaConverter.convert("{ }");
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testWrongType() {
+ JsonSchemaConverter.convert("{ type: 'whatever' }");
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testArrayWithAdditionalItems() {
+ JsonSchemaConverter.convert("{ type: 'array', items: [{type: 'integer'}], additionalItems: true }");
+ }
+
+ @Test
+ public void testMissingProperties() {
+ final TypeInformation<?> result = JsonSchemaConverter.convert("{ type: 'object' }");
+
+ assertEquals(Types.ROW(), result);
+ }
+
+ @Test
+ public void testNullUnionTypes() {
+ final TypeInformation<?> result = JsonSchemaConverter.convert("{ type: ['string', 'null'] }");
+
+ assertEquals(Types.STRING, result);
+ }
+
+ @Test
+ public void testTimestamp() {
+ final TypeInformation<?> result = JsonSchemaConverter.convert("{ type: 'string', format: 'date-time' }");
+
+ assertEquals(Types.SQL_TIMESTAMP, result);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-formats/flink-json/src/test/resources/complex-schema.json
----------------------------------------------------------------------
diff --git a/flink-formats/flink-json/src/test/resources/complex-schema.json b/flink-formats/flink-json/src/test/resources/complex-schema.json
new file mode 100644
index 0000000..8666428
--- /dev/null
+++ b/flink-formats/flink-json/src/test/resources/complex-schema.json
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+{
+ "$schema": "http://json-schema.org/draft-06/schema#",
+ "description": "A representation of a person, company, organization, or place",
+ "type": "object",
+ "required": [
+ "familyName",
+ "givenName"
+ ],
+ "properties": {
+ "fn": {
+ "description": "Formatted Name",
+ "type": "string"
+ },
+ "familyName": {
+ "type": "string"
+ },
+ "additionalName": {
+ "type": "boolean"
+ },
+ "tuples": {
+ "type": "array",
+ "items": [
+ {
+ "type": "number"
+ },
+ {
+ "type": "string"
+ },
+ {
+ "type": "string",
+ "enum": [
+ "Street",
+ "Avenue",
+ "Boulevard"
+ ]
+ },
+ {
+ "type": "string",
+ "enum": [
+ "NW",
+ "NE",
+ "SW",
+ "SE"
+ ]
+ }
+ ],
+ "additionalItems": false
+ },
+ "honorificPrefix": {
+ "type": "array",
+ "items": {
+ "type": "string"
+ }
+ },
+ "url": {
+ "type": "string",
+ "format": "uri"
+ },
+ "email": {
+ "type": "object",
+ "properties": {
+ "type": {
+ "type": "string"
+ },
+ "value": {
+ "type": "string",
+ "format": "email"
+ }
+ }
+ },
+ tel: {
+ "type": "object",
+ "properties": {
+ "type": {
+ "type": "integer"
+ },
+ "value": {
+ "type": "string",
+ "format": "phone"
+ }
+ }
+ },
+ "sound": {
+ "type": "null"
+ },
+ "org": {
+ "type": "object",
+ "properties": {
+ "organizationUnit": {
+ "type": "object",
+ "properties": {}
+ }
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-formats/flink-json/src/test/resources/reference-schema.json
----------------------------------------------------------------------
diff --git a/flink-formats/flink-json/src/test/resources/reference-schema.json b/flink-formats/flink-json/src/test/resources/reference-schema.json
new file mode 100644
index 0000000..99e0e79
--- /dev/null
+++ b/flink-formats/flink-json/src/test/resources/reference-schema.json
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+{
+ "$schema": "http://json-schema.org/draft-04/schema#",
+ "definitions": {
+ "address": {
+ "type": "object",
+ "properties": {
+ "street_address": {
+ "type": "string"
+ },
+ "city": {
+ "type": "string"
+ },
+ "state": {
+ "type": "string"
+ }
+ },
+ "required": [
+ "street_address",
+ "city",
+ "state"
+ ]
+ }
+ },
+ "type": "object",
+ "properties": {
+ "billing_address": {
+ "$ref": "#/definitions/address"
+ },
+ "shipping_address": {
+ "$ref": "#/definitions/address"
+ },
+ "optional_address": {
+ "oneOf": [
+ {
+ "type": "null"
+ },
+ {
+ "$ref": "#/definitions/address"
+ }
+ ]
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-formats/pom.xml
----------------------------------------------------------------------
diff --git a/flink-formats/pom.xml b/flink-formats/pom.xml
index d9dc3f1..ad776cf 100644
--- a/flink-formats/pom.xml
+++ b/flink-formats/pom.xml
@@ -37,6 +37,24 @@ under the License.
<modules>
<module>flink-avro</module>
+ <module>flink-json</module>
</modules>
-</project>
\ No newline at end of file
+ <!-- override these root dependencies as 'provided', so they don't end up
+ in the jars-with-dependencies (uber jars) of formats and
+ user programs that depend on the formats -->
+
+ <dependencies>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-test-utils-junit</artifactId>
+ </dependency>
+ </dependencies>
+
+</project>
[2/2] flink git commit: [FLINK-8630] [formats] Add proper support for
JSON formats
Posted by tw...@apache.org.
[FLINK-8630] [formats] Add proper support for JSON formats
This closes #5491.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/6c09c057
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/6c09c057
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/6c09c057
Branch: refs/heads/master
Commit: 6c09c057846fecd456bfbe77773dacabb4800370
Parents: 942649e
Author: twalthr <tw...@apache.org>
Authored: Mon Feb 12 18:18:44 2018 +0100
Committer: twalthr <tw...@apache.org>
Committed: Thu Feb 15 20:15:52 2018 +0100
----------------------------------------------------------------------
.../flink-connector-kafka-0.10/pom.xml | 16 +
.../kafka/Kafka010JsonTableSinkTest.java | 2 +-
.../kafka/Kafka010JsonTableSourceTest.java | 2 +-
.../flink-connector-kafka-0.11/pom.xml | 16 +
.../kafka/Kafka011JsonTableSourceTest.java | 2 +-
.../flink-connector-kafka-0.8/pom.xml | 16 +
.../kafka/Kafka08JsonTableSinkTest.java | 2 +-
.../kafka/Kafka08JsonTableSourceTest.java | 2 +-
.../flink-connector-kafka-0.9/pom.xml | 16 +
.../kafka/Kafka09JsonTableSinkTest.java | 2 +-
.../kafka/Kafka09JsonTableSourceTest.java | 2 +-
.../flink-connector-kafka-base/pom.xml | 16 +
.../connectors/kafka/KafkaJsonTableSink.java | 2 +-
.../connectors/kafka/KafkaJsonTableSource.java | 2 +-
.../JSONDeserializationSchema.java | 31 +-
.../JsonRowDeserializationSchema.java | 92 +----
.../JsonRowSerializationSchema.java | 65 +---
.../kafka/JSONDeserializationSchemaTest.java | 47 ---
.../kafka/JsonRowDeserializationSchemaTest.java | 131 -------
.../kafka/JsonRowSerializationSchemaTest.java | 127 -------
flink-dist/pom.xml | 16 +
flink-dist/src/main/assemblies/opt.xml | 15 +
flink-formats/flink-json/pom.xml | 81 +++++
.../json/JsonNodeDeserializationSchema.java | 47 +++
.../json/JsonRowDeserializationSchema.java | 217 +++++++++++
.../json/JsonRowSerializationSchema.java | 208 +++++++++++
.../flink/formats/json/JsonSchemaConverter.java | 361 +++++++++++++++++++
.../json/JsonNodeDeserializationSchemaTest.java | 47 +++
.../json/JsonRowDeserializationSchemaTest.java | 198 ++++++++++
.../json/JsonRowSerializationSchemaTest.java | 174 +++++++++
.../formats/json/JsonSchemaConverterTest.java | 114 ++++++
.../src/test/resources/complex-schema.json | 114 ++++++
.../src/test/resources/reference-schema.json | 61 ++++
flink-formats/pom.xml | 20 +-
34 files changed, 1789 insertions(+), 475 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-connectors/flink-connector-kafka-0.10/pom.xml
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-0.10/pom.xml b/flink-connectors/flink-connector-kafka-0.10/pom.xml
index 6e92d5f..8b4ff38 100644
--- a/flink-connectors/flink-connector-kafka-0.10/pom.xml
+++ b/flink-connectors/flink-connector-kafka-0.10/pom.xml
@@ -90,6 +90,14 @@ under the License.
<optional>true</optional>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-json</artifactId>
+ <version>${project.version}</version>
+ <!-- Projects depending on this project, won't depend on flink-json. -->
+ <optional>true</optional>
+ </dependency>
+
<!-- test dependencies -->
<dependency>
@@ -102,6 +110,14 @@ under the License.
<dependency>
<groupId>org.apache.flink</groupId>
+ <artifactId>flink-json</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
<version>${project.version}</version>
<scope>test</scope>
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSinkTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSinkTest.java b/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSinkTest.java
index 4d805d5..af562c6 100644
--- a/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSinkTest.java
+++ b/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSinkTest.java
@@ -19,8 +19,8 @@
package org.apache.flink.streaming.connectors.kafka;
import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.formats.json.JsonRowSerializationSchema;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
-import org.apache.flink.streaming.util.serialization.JsonRowSerializationSchema;
import org.apache.flink.types.Row;
import java.util.Properties;
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSourceTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSourceTest.java b/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSourceTest.java
index 0f6c531..f9d4d84 100644
--- a/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSourceTest.java
+++ b/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSourceTest.java
@@ -19,7 +19,7 @@
package org.apache.flink.streaming.connectors.kafka;
import org.apache.flink.api.common.serialization.DeserializationSchema;
-import org.apache.flink.streaming.util.serialization.JsonRowDeserializationSchema;
+import org.apache.flink.formats.json.JsonRowDeserializationSchema;
import org.apache.flink.types.Row;
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-connectors/flink-connector-kafka-0.11/pom.xml
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-0.11/pom.xml b/flink-connectors/flink-connector-kafka-0.11/pom.xml
index 697394d..1e935f6 100644
--- a/flink-connectors/flink-connector-kafka-0.11/pom.xml
+++ b/flink-connectors/flink-connector-kafka-0.11/pom.xml
@@ -90,6 +90,14 @@ under the License.
<optional>true</optional>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-json</artifactId>
+ <version>${project.version}</version>
+ <!-- Projects depending on this project, won't depend on flink-json. -->
+ <optional>true</optional>
+ </dependency>
+
<!-- test dependencies -->
<dependency>
@@ -111,6 +119,14 @@ under the License.
<dependency>
<groupId>org.apache.flink</groupId>
+ <artifactId>flink-json</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
<version>${project.version}</version>
<scope>test</scope>
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011JsonTableSourceTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011JsonTableSourceTest.java b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011JsonTableSourceTest.java
index fc2957e..b35d1c5 100644
--- a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011JsonTableSourceTest.java
+++ b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011JsonTableSourceTest.java
@@ -19,7 +19,7 @@
package org.apache.flink.streaming.connectors.kafka;
import org.apache.flink.api.common.serialization.DeserializationSchema;
-import org.apache.flink.streaming.util.serialization.JsonRowDeserializationSchema;
+import org.apache.flink.formats.json.JsonRowDeserializationSchema;
import org.apache.flink.types.Row;
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-connectors/flink-connector-kafka-0.8/pom.xml
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-0.8/pom.xml b/flink-connectors/flink-connector-kafka-0.8/pom.xml
index 21a831a..a58591e 100644
--- a/flink-connectors/flink-connector-kafka-0.8/pom.xml
+++ b/flink-connectors/flink-connector-kafka-0.8/pom.xml
@@ -82,6 +82,14 @@ under the License.
</dependency>
<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-json</artifactId>
+ <version>${project.version}</version>
+ <!-- Projects depending on this project, won't depend on flink-json. -->
+ <optional>true</optional>
+ </dependency>
+
+ <dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_${scala.binary.version}</artifactId>
<version>${kafka.version}</version>
@@ -144,6 +152,14 @@ under the License.
</dependency>
<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-json</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
+
+ <dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-test</artifactId>
<version>${curator.version}</version>
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSinkTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSinkTest.java b/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSinkTest.java
index d7bb683..53da9f6 100644
--- a/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSinkTest.java
+++ b/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSinkTest.java
@@ -19,8 +19,8 @@
package org.apache.flink.streaming.connectors.kafka;
import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.formats.json.JsonRowSerializationSchema;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
-import org.apache.flink.streaming.util.serialization.JsonRowSerializationSchema;
import org.apache.flink.types.Row;
import java.util.Properties;
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSourceTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSourceTest.java b/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSourceTest.java
index 7e3349c..b66e8ff 100644
--- a/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSourceTest.java
+++ b/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSourceTest.java
@@ -19,7 +19,7 @@
package org.apache.flink.streaming.connectors.kafka;
import org.apache.flink.api.common.serialization.DeserializationSchema;
-import org.apache.flink.streaming.util.serialization.JsonRowDeserializationSchema;
+import org.apache.flink.formats.json.JsonRowDeserializationSchema;
import org.apache.flink.types.Row;
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-connectors/flink-connector-kafka-0.9/pom.xml
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-0.9/pom.xml b/flink-connectors/flink-connector-kafka-0.9/pom.xml
index d976dce..d07cb5a 100644
--- a/flink-connectors/flink-connector-kafka-0.9/pom.xml
+++ b/flink-connectors/flink-connector-kafka-0.9/pom.xml
@@ -81,6 +81,14 @@ under the License.
</dependency>
<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-json</artifactId>
+ <version>${project.version}</version>
+ <!-- Projects depending on this project, won't depend on flink-json. -->
+ <optional>true</optional>
+ </dependency>
+
+ <dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>${kafka.version}</version>
@@ -98,6 +106,14 @@ under the License.
<dependency>
<groupId>org.apache.flink</groupId>
+ <artifactId>flink-json</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
<version>${project.version}</version>
<scope>test</scope>
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09JsonTableSinkTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09JsonTableSinkTest.java b/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09JsonTableSinkTest.java
index 58f2b05..610e048 100644
--- a/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09JsonTableSinkTest.java
+++ b/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09JsonTableSinkTest.java
@@ -19,8 +19,8 @@
package org.apache.flink.streaming.connectors.kafka;
import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.formats.json.JsonRowSerializationSchema;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
-import org.apache.flink.streaming.util.serialization.JsonRowSerializationSchema;
import org.apache.flink.types.Row;
import java.util.Properties;
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09JsonTableSourceTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09JsonTableSourceTest.java b/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09JsonTableSourceTest.java
index aeee175..65fa0f1 100644
--- a/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09JsonTableSourceTest.java
+++ b/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09JsonTableSourceTest.java
@@ -19,7 +19,7 @@
package org.apache.flink.streaming.connectors.kafka;
import org.apache.flink.api.common.serialization.DeserializationSchema;
-import org.apache.flink.streaming.util.serialization.JsonRowDeserializationSchema;
+import org.apache.flink.formats.json.JsonRowDeserializationSchema;
import org.apache.flink.types.Row;
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-connectors/flink-connector-kafka-base/pom.xml
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-base/pom.xml b/flink-connectors/flink-connector-kafka-base/pom.xml
index 7330c54..e7412cf 100644
--- a/flink-connectors/flink-connector-kafka-base/pom.xml
+++ b/flink-connectors/flink-connector-kafka-base/pom.xml
@@ -75,6 +75,14 @@ under the License.
</dependency>
<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-json</artifactId>
+ <version>${project.version}</version>
+ <!-- Projects depending on this project, won't depend on flink-json. -->
+ <optional>true</optional>
+ </dependency>
+
+ <dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_${scala.binary.version}</artifactId>
<version>${kafka.version}</version>
@@ -183,6 +191,14 @@ under the License.
</dependency>
<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-json</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-minikdc</artifactId>
<version>${minikdc.version}</version>
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSink.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSink.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSink.java
index bfc2143..ec27398 100644
--- a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSink.java
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSink.java
@@ -21,8 +21,8 @@ package org.apache.flink.streaming.connectors.kafka;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.formats.json.JsonRowSerializationSchema;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
-import org.apache.flink.streaming.util.serialization.JsonRowSerializationSchema;
import org.apache.flink.types.Row;
import java.util.Properties;
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSource.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSource.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSource.java
index b3b545e..f581e89 100644
--- a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSource.java
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSource.java
@@ -21,7 +21,7 @@ package org.apache.flink.streaming.connectors.kafka;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
-import org.apache.flink.streaming.util.serialization.JsonRowDeserializationSchema;
+import org.apache.flink.formats.json.JsonRowDeserializationSchema;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.sources.DefinedFieldMapping;
import org.apache.flink.table.sources.StreamTableSource;
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/util/serialization/JSONDeserializationSchema.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/util/serialization/JSONDeserializationSchema.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/util/serialization/JSONDeserializationSchema.java
index 900c094..9d55473 100644
--- a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/util/serialization/JSONDeserializationSchema.java
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/util/serialization/JSONDeserializationSchema.java
@@ -18,36 +18,17 @@
package org.apache.flink.streaming.util.serialization;
import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.api.common.serialization.AbstractDeserializationSchema;
-
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
-
-import java.io.IOException;
+import org.apache.flink.formats.json.JsonNodeDeserializationSchema;
/**
* DeserializationSchema that deserializes a JSON String into an ObjectNode.
*
* <p>Fields can be accessed by calling objectNode.get(<name>).as(<type>)
+ *
+ * @deprecated Please use {@link JsonNodeDeserializationSchema} in the "flink-json" module.
*/
@PublicEvolving
-public class JSONDeserializationSchema extends AbstractDeserializationSchema<ObjectNode> {
-
- private static final long serialVersionUID = -1699854177598621044L;
-
- private ObjectMapper mapper;
-
- @Override
- public ObjectNode deserialize(byte[] message) throws IOException {
- if (mapper == null) {
- mapper = new ObjectMapper();
- }
- return mapper.readValue(message, ObjectNode.class);
- }
-
- @Override
- public boolean isEndOfStream(ObjectNode nextElement) {
- return false;
- }
-
+@Deprecated
+public class JSONDeserializationSchema extends JsonNodeDeserializationSchema {
+ // delegate everything to the parent class
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/util/serialization/JsonRowDeserializationSchema.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/util/serialization/JsonRowDeserializationSchema.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/util/serialization/JsonRowDeserializationSchema.java
index 21215cd..a09f103 100644
--- a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/util/serialization/JsonRowDeserializationSchema.java
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/util/serialization/JsonRowDeserializationSchema.java
@@ -19,16 +19,8 @@
package org.apache.flink.streaming.util.serialization;
import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.types.Row;
-import org.apache.flink.util.Preconditions;
-
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
-
-import java.io.IOException;
/**
* Deserialization schema from JSON to {@link Row}.
@@ -37,89 +29,25 @@ import java.io.IOException;
* the specified fields.
*
* <p>Failure during deserialization are forwarded as wrapped IOExceptions.
+ *
+ * @deprecated Please use {@link org.apache.flink.formats.json.JsonRowDeserializationSchema} in
+ * the "flink-json" module.
*/
@PublicEvolving
-public class JsonRowDeserializationSchema implements DeserializationSchema<Row> {
-
- private static final long serialVersionUID = -228294330688809195L;
-
- /** Type information describing the result type. */
- private final TypeInformation<Row> typeInfo;
-
- /** Field names to parse. Indices match fieldTypes indices. */
- private final String[] fieldNames;
-
- /** Types to parse fields as. Indices match fieldNames indices. */
- private final TypeInformation<?>[] fieldTypes;
-
- /** Object mapper for parsing the JSON. */
- private final ObjectMapper objectMapper = new ObjectMapper();
-
- /** Flag indicating whether to fail on a missing field. */
- private boolean failOnMissingField;
+@Deprecated
+public class JsonRowDeserializationSchema extends org.apache.flink.formats.json.JsonRowDeserializationSchema {
/**
* Creates a JSON deserialization schema for the given fields and types.
*
* @param typeInfo Type information describing the result type. The field names are used
* to parse the JSON file and so are the types.
- */
- public JsonRowDeserializationSchema(TypeInformation<Row> typeInfo) {
- Preconditions.checkNotNull(typeInfo, "Type information");
- this.typeInfo = typeInfo;
-
- this.fieldNames = ((RowTypeInfo) typeInfo).getFieldNames();
- this.fieldTypes = ((RowTypeInfo) typeInfo).getFieldTypes();
- }
-
- @Override
- public Row deserialize(byte[] message) throws IOException {
- try {
- JsonNode root = objectMapper.readTree(message);
-
- Row row = new Row(fieldNames.length);
- for (int i = 0; i < fieldNames.length; i++) {
- JsonNode node = root.get(fieldNames[i]);
-
- if (node == null) {
- if (failOnMissingField) {
- throw new IllegalStateException("Failed to find field with name '"
- + fieldNames[i] + "'.");
- } else {
- row.setField(i, null);
- }
- } else {
- // Read the value as specified type
- Object value = objectMapper.treeToValue(node, fieldTypes[i].getTypeClass());
- row.setField(i, value);
- }
- }
-
- return row;
- } catch (Throwable t) {
- throw new IOException("Failed to deserialize JSON object.", t);
- }
- }
-
- @Override
- public boolean isEndOfStream(Row nextElement) {
- return false;
- }
-
- @Override
- public TypeInformation<Row> getProducedType() {
- return typeInfo;
- }
-
- /**
- * Configures the failure behaviour if a JSON field is missing.
*
- * <p>By default, a missing field is ignored and the field is set to null.
- *
- * @param failOnMissingField Flag indicating whether to fail or not on a missing field.
+ * @deprecated Please use {@link org.apache.flink.formats.json.JsonRowDeserializationSchema} in
+ * the "flink-json" module.
*/
- public void setFailOnMissingField(boolean failOnMissingField) {
- this.failOnMissingField = failOnMissingField;
+ @Deprecated
+ public JsonRowDeserializationSchema(TypeInformation<Row> typeInfo) {
+ super(typeInfo);
}
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/util/serialization/JsonRowSerializationSchema.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/util/serialization/JsonRowSerializationSchema.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/util/serialization/JsonRowSerializationSchema.java
index 44ce7ea..83f7519 100644
--- a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/util/serialization/JsonRowSerializationSchema.java
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/util/serialization/JsonRowSerializationSchema.java
@@ -18,16 +18,8 @@
package org.apache.flink.streaming.util.serialization;
import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.common.typeutils.CompositeType;
-import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.types.Row;
-import org.apache.flink.util.Preconditions;
-
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
/**
* Serialization schema that serializes an object into a JSON bytes.
@@ -37,57 +29,24 @@ import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.Obje
*
* <p>Result <code>byte[]</code> messages can be deserialized using
* {@link JsonRowDeserializationSchema}.
+ *
+ * @deprecated Please use {@link org.apache.flink.formats.json.JsonRowSerializationSchema} in
+ * the "flink-json" module.
*/
@PublicEvolving
-public class JsonRowSerializationSchema implements SerializationSchema<Row> {
-
- private static final long serialVersionUID = -2885556750743978636L;
-
- /** Fields names in the input Row object. */
- private final String[] fieldNames;
- /** Object mapper that is used to create output JSON objects. */
- private static ObjectMapper mapper = new ObjectMapper();
+@Deprecated
+public class JsonRowSerializationSchema extends org.apache.flink.formats.json.JsonRowSerializationSchema {
/**
* Creates a JSON serialization schema for the given fields and types.
*
- * @param rowSchema The schema of the rows to encode.
+ * @param typeInfo The schema of the rows to encode.
+ *
+ * @deprecated Please use {@link org.apache.flink.formats.json.JsonRowSerializationSchema} in
+ * the "flink-json" module.
*/
- public JsonRowSerializationSchema(RowTypeInfo rowSchema) {
-
- Preconditions.checkNotNull(rowSchema);
- String[] fieldNames = rowSchema.getFieldNames();
- TypeInformation[] fieldTypes = rowSchema.getFieldTypes();
-
- // check that no field is composite
- for (int i = 0; i < fieldTypes.length; i++) {
- if (fieldTypes[i] instanceof CompositeType) {
- throw new IllegalArgumentException("JsonRowSerializationSchema cannot encode rows with nested schema, " +
- "but field '" + fieldNames[i] + "' is nested: " + fieldTypes[i].toString());
- }
- }
-
- this.fieldNames = fieldNames;
- }
-
- @Override
- public byte[] serialize(Row row) {
- if (row.getArity() != fieldNames.length) {
- throw new IllegalStateException(String.format(
- "Number of elements in the row %s is different from number of field names: %d", row, fieldNames.length));
- }
-
- ObjectNode objectNode = mapper.createObjectNode();
-
- for (int i = 0; i < row.getArity(); i++) {
- JsonNode node = mapper.valueToTree(row.getField(i));
- objectNode.set(fieldNames[i], node);
- }
-
- try {
- return mapper.writeValueAsBytes(objectNode);
- } catch (Exception e) {
- throw new RuntimeException("Failed to serialize row", e);
- }
+ @Deprecated
+ public JsonRowSerializationSchema(TypeInformation<Row> typeInfo) {
+ super(typeInfo);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/JSONDeserializationSchemaTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/JSONDeserializationSchemaTest.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/JSONDeserializationSchemaTest.java
deleted file mode 100644
index 4bb420b..0000000
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/JSONDeserializationSchemaTest.java
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.streaming.connectors.kafka;
-
-import org.apache.flink.streaming.util.serialization.JSONDeserializationSchema;
-
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
-
-import org.junit.Assert;
-import org.junit.Test;
-
-import java.io.IOException;
-
-/**
- * Tests for the {@link JSONDeserializationSchema}.
- */
-public class JSONDeserializationSchemaTest {
- @Test
- public void testDeserialize() throws IOException {
- ObjectMapper mapper = new ObjectMapper();
- ObjectNode initialValue = mapper.createObjectNode();
- initialValue.put("key", 4).put("value", "world");
- byte[] serializedValue = mapper.writeValueAsBytes(initialValue);
-
- JSONDeserializationSchema schema = new JSONDeserializationSchema();
- ObjectNode deserializedValue = schema.deserialize(serializedValue);
-
- Assert.assertEquals(4, deserializedValue.get("key").asInt());
- Assert.assertEquals("world", deserializedValue.get("value").asText());
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/JsonRowDeserializationSchemaTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/JsonRowDeserializationSchemaTest.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/JsonRowDeserializationSchemaTest.java
deleted file mode 100644
index a5e6e7e..0000000
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/JsonRowDeserializationSchemaTest.java
+++ /dev/null
@@ -1,131 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.streaming.connectors.kafka;
-
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.streaming.util.serialization.JsonRowDeserializationSchema;
-import org.apache.flink.table.api.Types;
-import org.apache.flink.types.Row;
-
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
-
-import org.junit.Test;
-
-import java.io.IOException;
-import java.util.concurrent.ThreadLocalRandom;
-
-import static org.junit.Assert.assertArrayEquals;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-
-/**
- * Tests for the {@link JsonRowDeserializationSchema}.
- */
-public class JsonRowDeserializationSchemaTest {
-
- /**
- * Tests simple deserialization.
- */
- @Test
- public void testDeserialization() throws Exception {
- long id = 1238123899121L;
- String name = "asdlkjasjkdla998y1122";
- byte[] bytes = new byte[1024];
- ThreadLocalRandom.current().nextBytes(bytes);
-
- ObjectMapper objectMapper = new ObjectMapper();
-
- // Root
- ObjectNode root = objectMapper.createObjectNode();
- root.put("id", id);
- root.put("name", name);
- root.put("bytes", bytes);
-
- byte[] serializedJson = objectMapper.writeValueAsBytes(root);
-
- JsonRowDeserializationSchema deserializationSchema = new JsonRowDeserializationSchema(
- Types.ROW(
- new String[] { "id", "name", "bytes" },
- new TypeInformation<?>[] { Types.LONG(), Types.STRING(), Types.PRIMITIVE_ARRAY(Types.BYTE()) }
- )
- );
-
- Row deserialized = deserializationSchema.deserialize(serializedJson);
-
- assertEquals(3, deserialized.getArity());
- assertEquals(id, deserialized.getField(0));
- assertEquals(name, deserialized.getField(1));
- assertArrayEquals(bytes, (byte[]) deserialized.getField(2));
- }
-
- /**
- * Tests deserialization with non-existing field name.
- */
- @Test
- public void testMissingNode() throws Exception {
- ObjectMapper objectMapper = new ObjectMapper();
-
- // Root
- ObjectNode root = objectMapper.createObjectNode();
- root.put("id", 123123123);
- byte[] serializedJson = objectMapper.writeValueAsBytes(root);
-
- JsonRowDeserializationSchema deserializationSchema = new JsonRowDeserializationSchema(
- Types.ROW(
- new String[] { "name" },
- new TypeInformation<?>[] { Types.STRING() }
- )
- );
-
- Row row = deserializationSchema.deserialize(serializedJson);
-
- assertEquals(1, row.getArity());
- assertNull("Missing field not null", row.getField(0));
-
- deserializationSchema.setFailOnMissingField(true);
-
- try {
- deserializationSchema.deserialize(serializedJson);
- fail("Did not throw expected Exception");
- } catch (IOException e) {
- assertTrue(e.getCause() instanceof IllegalStateException);
- }
- }
-
- /**
- * Tests that number of field names and types has to match.
- */
- @Test
- public void testNumberOfFieldNamesAndTypesMismatch() throws Exception {
- try {
- new JsonRowDeserializationSchema(
- Types.ROW(
- new String[] { "one", "two", "three" },
- new TypeInformation<?>[] { Types.LONG() }
- )
- );
- fail("Did not throw expected Exception");
- } catch (IllegalArgumentException ignored) {
- // Expected
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/JsonRowSerializationSchemaTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/JsonRowSerializationSchemaTest.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/JsonRowSerializationSchemaTest.java
deleted file mode 100644
index 70140a6..0000000
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/JsonRowSerializationSchemaTest.java
+++ /dev/null
@@ -1,127 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.streaming.connectors.kafka;
-
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.typeutils.RowTypeInfo;
-import org.apache.flink.streaming.util.serialization.JsonRowDeserializationSchema;
-import org.apache.flink.streaming.util.serialization.JsonRowSerializationSchema;
-import org.apache.flink.table.api.Types;
-import org.apache.flink.types.Row;
-
-import org.junit.Test;
-
-import java.io.IOException;
-
-import static org.junit.Assert.assertEquals;
-
-/**
- * Tests for the {@link JsonRowSerializationSchema}.
- */
-public class JsonRowSerializationSchemaTest {
-
- @Test
- public void testRowSerialization() throws IOException {
- RowTypeInfo rowSchema = new RowTypeInfo(
- new TypeInformation[]{Types.INT(), Types.BOOLEAN(), Types.STRING()},
- new String[] {"f1", "f2", "f3"}
- );
-
- Row row = new Row(3);
- row.setField(0, 1);
- row.setField(1, true);
- row.setField(2, "str");
-
- Row resultRow = serializeAndDeserialize(rowSchema, row);
- assertEqualRows(row, resultRow);
- }
-
- @Test
- public void testSerializationOfTwoRows() throws IOException {
- RowTypeInfo rowSchema = new RowTypeInfo(
- new TypeInformation[]{Types.INT(), Types.BOOLEAN(), Types.STRING()},
- new String[] {"f1", "f2", "f3"}
- );
-
- Row row1 = new Row(3);
- row1.setField(0, 1);
- row1.setField(1, true);
- row1.setField(2, "str");
-
- JsonRowSerializationSchema serializationSchema = new JsonRowSerializationSchema(rowSchema);
- JsonRowDeserializationSchema deserializationSchema = new JsonRowDeserializationSchema(rowSchema);
-
- byte[] bytes = serializationSchema.serialize(row1);
- assertEqualRows(row1, deserializationSchema.deserialize(bytes));
-
- Row row2 = new Row(3);
- row2.setField(0, 10);
- row2.setField(1, false);
- row2.setField(2, "newStr");
-
- bytes = serializationSchema.serialize(row2);
- assertEqualRows(row2, deserializationSchema.deserialize(bytes));
- }
-
- @Test(expected = NullPointerException.class)
- public void testInputValidation() {
- new JsonRowSerializationSchema(null);
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void testRejectNestedSchema() {
- RowTypeInfo rowSchema = new RowTypeInfo(
- new TypeInformation[]{Types.INT(), Types.BOOLEAN(), Types.ROW(Types.INT(), Types.DOUBLE())},
- new String[] {"f1", "f2", "f3"}
- );
-
- new JsonRowSerializationSchema(rowSchema);
- }
-
- @Test(expected = IllegalStateException.class)
- public void testSerializeRowWithInvalidNumberOfFields() {
- RowTypeInfo rowSchema = new RowTypeInfo(
- new TypeInformation[]{Types.INT(), Types.BOOLEAN(), Types.STRING()},
- new String[] {"f1", "f2", "f3"}
- );
-
- Row row = new Row(1);
- row.setField(0, 1);
-
- JsonRowSerializationSchema serializationSchema = new JsonRowSerializationSchema(rowSchema);
- serializationSchema.serialize(row);
- }
-
- private Row serializeAndDeserialize(RowTypeInfo rowSchema, Row row) throws IOException {
- JsonRowSerializationSchema serializationSchema = new JsonRowSerializationSchema(rowSchema);
- JsonRowDeserializationSchema deserializationSchema = new JsonRowDeserializationSchema(rowSchema);
-
- byte[] bytes = serializationSchema.serialize(row);
- return deserializationSchema.deserialize(bytes);
- }
-
- private void assertEqualRows(Row expectedRow, Row resultRow) {
- assertEquals("Deserialized row should have expected number of fields",
- expectedRow.getArity(), resultRow.getArity());
- for (int i = 0; i < expectedRow.getArity(); i++) {
- assertEquals(String.format("Field number %d should be as in the original row", i),
- expectedRow.getField(i), resultRow.getField(i));
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-dist/pom.xml
----------------------------------------------------------------------
diff --git a/flink-dist/pom.xml b/flink-dist/pom.xml
index c85aad2..daaaa82 100644
--- a/flink-dist/pom.xml
+++ b/flink-dist/pom.xml
@@ -330,6 +330,22 @@ under the License.
</dependency>
<!-- end optional Flink libraries -->
+ <!-- start optional Flink formats -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-avro</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-json</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <!-- end optional Flink formats -->
+
<!-- test dependencies -->
<dependency>
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-dist/src/main/assemblies/opt.xml
----------------------------------------------------------------------
diff --git a/flink-dist/src/main/assemblies/opt.xml b/flink-dist/src/main/assemblies/opt.xml
index 3f724e3..86f6e01 100644
--- a/flink-dist/src/main/assemblies/opt.xml
+++ b/flink-dist/src/main/assemblies/opt.xml
@@ -153,5 +153,20 @@
<destName>flink-queryable-state-runtime_${scala.binary.version}-${project.version}.jar</destName>
<fileMode>0644</fileMode>
</file>
+
+ <!-- Formats -->
+ <file>
+ <source>../flink-formats/flink-avro/target/flink-avro-${project.version}.jar</source>
+ <outputDirectory>opt/</outputDirectory>
+ <destName>flink-avro-${project.version}.jar</destName>
+ <fileMode>0644</fileMode>
+ </file>
+
+ <file>
+ <source>../flink-formats/flink-json/target/flink-json-${project.version}.jar</source>
+ <outputDirectory>opt/</outputDirectory>
+ <destName>flink-json-${project.version}.jar</destName>
+ <fileMode>0644</fileMode>
+ </file>
</files>
</assembly>
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-formats/flink-json/pom.xml
----------------------------------------------------------------------
diff --git a/flink-formats/flink-json/pom.xml b/flink-formats/flink-json/pom.xml
new file mode 100644
index 0000000..d51f4fa
--- /dev/null
+++ b/flink-formats/flink-json/pom.xml
@@ -0,0 +1,81 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-formats</artifactId>
+ <version>1.5-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <artifactId>flink-json</artifactId>
+ <name>flink-json</name>
+
+ <packaging>jar</packaging>
+
+ <dependencies>
+
+ <!-- core dependencies -->
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-shaded-jackson</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-core</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- test dependencies -->
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-core</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ <executions>
+ <execution>
+ <goals>
+ <goal>test-jar</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
+</project>
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonNodeDeserializationSchema.java
----------------------------------------------------------------------
diff --git a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonNodeDeserializationSchema.java b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonNodeDeserializationSchema.java
new file mode 100644
index 0000000..7501cc3
--- /dev/null
+++ b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonNodeDeserializationSchema.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.formats.json;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.serialization.AbstractDeserializationSchema;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
+
+import java.io.IOException;
+
+/**
+ * DeserializationSchema that deserializes a JSON String into an ObjectNode.
+ *
+ * <p>Fields can be accessed by calling objectNode.get(<name>).as(<type>)
+ */
+@PublicEvolving
+public class JsonNodeDeserializationSchema extends AbstractDeserializationSchema<ObjectNode> {
+
+ private static final long serialVersionUID = -1699854177598621044L;
+
+ private ObjectMapper mapper;
+
+ @Override
+ public ObjectNode deserialize(byte[] message) throws IOException {
+ if (mapper == null) {
+ mapper = new ObjectMapper();
+ }
+ return mapper.readValue(message, ObjectNode.class);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowDeserializationSchema.java
----------------------------------------------------------------------
diff --git a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowDeserializationSchema.java b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowDeserializationSchema.java
new file mode 100644
index 0000000..edc4b01
--- /dev/null
+++ b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowDeserializationSchema.java
@@ -0,0 +1,217 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.formats.json;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+
+import java.io.IOException;
+import java.lang.reflect.Array;
+import java.sql.Date;
+import java.sql.Time;
+import java.sql.Timestamp;
+
+/**
+ * Deserialization schema from JSON to Flink types.
+ *
+ * <p>Deserializes a <code>byte[]</code> message as a JSON object and reads
+ * the specified fields.
+ *
+ * <p>Failure during deserialization are forwarded as wrapped IOExceptions.
+ */
+@PublicEvolving
+public class JsonRowDeserializationSchema implements DeserializationSchema<Row> {
+
+ private static final long serialVersionUID = -228294330688809195L;
+
+ /** Type information describing the result type. */
+ private final TypeInformation<Row> typeInfo;
+
+ /** Object mapper for parsing the JSON. */
+ private final ObjectMapper objectMapper = new ObjectMapper();
+
+ /** Flag indicating whether to fail on a missing field. */
+ private boolean failOnMissingField;
+
+ /**
+ * Creates a JSON deserialization schema for the given type information.
+ *
+ * @param typeInfo Type information describing the result type. The field names of {@link Row}
+ * are used to parse the JSON properties.
+ */
+ public JsonRowDeserializationSchema(TypeInformation<Row> typeInfo) {
+ Preconditions.checkNotNull(typeInfo, "Type information");
+ this.typeInfo = typeInfo;
+
+ if (!(typeInfo instanceof RowTypeInfo)) {
+ throw new IllegalArgumentException("Row type information expected.");
+ }
+ }
+
+ /**
+ * Creates a JSON deserialization schema for the given JSON schema.
+ *
+ * @param jsonSchema JSON schema describing the result type
+ *
+ * @see <a href="http://json-schema.org/">http://json-schema.org/</a>
+ */
+ public JsonRowDeserializationSchema(String jsonSchema) {
+ this(JsonSchemaConverter.convert(jsonSchema));
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public Row deserialize(byte[] message) throws IOException {
+ try {
+ final JsonNode root = objectMapper.readTree(message);
+ return convertRow(root, (RowTypeInfo) typeInfo);
+ } catch (Throwable t) {
+ throw new IOException("Failed to deserialize JSON object.", t);
+ }
+ }
+
+ @Override
+ public boolean isEndOfStream(Row nextElement) {
+ return false;
+ }
+
+ @Override
+ public TypeInformation<Row> getProducedType() {
+ return typeInfo;
+ }
+
+ /**
+ * Configures the failure behaviour if a JSON field is missing.
+ *
+ * <p>By default, a missing field is ignored and the field is set to null.
+ *
+ * @param failOnMissingField Flag indicating whether to fail or not on a missing field.
+ */
+ public void setFailOnMissingField(boolean failOnMissingField) {
+ this.failOnMissingField = failOnMissingField;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ private Object convert(JsonNode node, TypeInformation<?> info) {
+ if (info == Types.VOID || node.isNull()) {
+ return null;
+ } else if (info == Types.BOOLEAN) {
+ return node.asBoolean();
+ } else if (info == Types.STRING) {
+ return node.asText();
+ } else if (info == Types.BIG_DEC) {
+ return node.decimalValue();
+ } else if (info == Types.BIG_INT) {
+ return node.bigIntegerValue();
+ } else if (info == Types.SQL_DATE) {
+ return Date.valueOf(node.asText());
+ } else if (info == Types.SQL_TIME) {
+ // according to RFC 3339 every full-time must have a timezone;
+ // until we have full timezone support, we only support UTC;
+ // users can parse their time as string as a workaround
+ final String time = node.asText();
+ if (time.indexOf('Z') < 0 || time.indexOf('.') >= 0) {
+ throw new IllegalStateException(
+ "Invalid time format. Only a time in UTC timezone without milliseconds is supported yet. " +
+ "Format: HH:mm:ss'Z'");
+ }
+ return Time.valueOf(time.substring(0, time.length() - 1));
+ } else if (info == Types.SQL_TIMESTAMP) {
+ // according to RFC 3339 every date-time must have a timezone;
+ // until we have full timezone support, we only support UTC;
+ // users can parse their time as string as a workaround
+ final String timestamp = node.asText();
+ if (timestamp.indexOf('Z') < 0) {
+ throw new IllegalStateException(
+ "Invalid timestamp format. Only a timestamp in UTC timezone is supported yet. " +
+ "Format: yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
+ }
+ return Timestamp.valueOf(timestamp.substring(0, timestamp.length() - 1).replace('T', ' '));
+ } else if (info instanceof RowTypeInfo) {
+ return convertRow(node, (RowTypeInfo) info);
+ } else if (info instanceof ObjectArrayTypeInfo) {
+ return convertObjectArray(node, ((ObjectArrayTypeInfo) info).getComponentInfo());
+ } else if (info instanceof BasicArrayTypeInfo) {
+ return convertObjectArray(node, ((BasicArrayTypeInfo) info).getComponentInfo());
+ } else if (info instanceof PrimitiveArrayTypeInfo &&
+ ((PrimitiveArrayTypeInfo) info).getComponentType() == Types.BYTE) {
+ return convertByteArray(node);
+ } else {
+ // for types that were specified without JSON schema
+ // e.g. POJOs
+ try {
+ return objectMapper.treeToValue(node, info.getTypeClass());
+ } catch (JsonProcessingException e) {
+ throw new IllegalStateException("Unsupported type information '" + info + "' for node: " + node);
+ }
+ }
+ }
+
+ private Row convertRow(JsonNode node, RowTypeInfo info) {
+ final String[] names = info.getFieldNames();
+ final TypeInformation<?>[] types = info.getFieldTypes();
+
+ final Row row = new Row(names.length);
+ for (int i = 0; i < names.length; i++) {
+ final String name = names[i];
+ final JsonNode subNode = node.get(name);
+ if (subNode == null) {
+ if (failOnMissingField) {
+ throw new IllegalStateException(
+ "Could not find field with name '" + name + "'.");
+ } else {
+ row.setField(i, null);
+ }
+ } else {
+ row.setField(i, convert(subNode, types[i]));
+ }
+ }
+
+ return row;
+ }
+
+ private Object convertObjectArray(JsonNode node, TypeInformation<?> elementType) {
+ final Object[] array = (Object[]) Array.newInstance(elementType.getTypeClass(), node.size());
+ for (int i = 0; i < node.size(); i++) {
+ array[i] = convert(node.get(i), elementType);
+ }
+ return array;
+ }
+
+ private Object convertByteArray(JsonNode node) {
+ try {
+ return node.binaryValue();
+ } catch (IOException e) {
+ throw new RuntimeException("Unable to deserialize byte array.", e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowSerializationSchema.java
----------------------------------------------------------------------
diff --git a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowSerializationSchema.java b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowSerializationSchema.java
new file mode 100644
index 0000000..8fee6a4
--- /dev/null
+++ b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowSerializationSchema.java
@@ -0,0 +1,208 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.formats.json;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ArrayNode;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ContainerNode;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.sql.Time;
+import java.sql.Timestamp;
+import java.text.SimpleDateFormat;
+
+/**
+ * Serialization schema that serializes an object of Flink types into a JSON bytes.
+ *
+ * <p>Serializes the input Flink object into a JSON string and
+ * converts it into <code>byte[]</code>.
+ *
+ * <p>Result <code>byte[]</code> messages can be deserialized using {@link JsonRowDeserializationSchema}.
+ */
+@PublicEvolving
+public class JsonRowSerializationSchema implements SerializationSchema<Row> {
+
+ private static final long serialVersionUID = -2885556750743978636L;
+
+ /** Type information describing the input type. */
+ private final TypeInformation<Row> typeInfo;
+
+ /** Object mapper that is used to create output JSON objects. */
+ private final ObjectMapper mapper = new ObjectMapper();
+
+ /** Formatter for RFC 3339-compliant string representation of a time value (with UTC timezone, without milliseconds). */
+ private SimpleDateFormat timeFormat = new SimpleDateFormat("HH:mm:ss'Z'");
+
+ /** Formatter for RFC 3339-compliant string representation of a time value (with UTC timezone). */
+ private SimpleDateFormat timeFormatWithMillis = new SimpleDateFormat("HH:mm:ss.SSS'Z'");
+
+ /** Formatter for RFC 3339-compliant string representation of a timestamp value (with UTC timezone). */
+ private SimpleDateFormat timestampFormat = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
+
+ /** Reusable object node. */
+ private transient ObjectNode node;
+
+ /**
+ * Creates a JSON serialization schema for the given type information.
+ *
+ * @param typeInfo The field names of {@link Row} are used to map to JSON properties.
+ */
+ public JsonRowSerializationSchema(TypeInformation<Row> typeInfo) {
+ Preconditions.checkNotNull(typeInfo, "Type information");
+ this.typeInfo = typeInfo;
+ }
+
+ /**
+ * Creates a JSON serialization schema for the given JSON schema.
+ *
+ * @param jsonSchema JSON schema describing the result type
+ *
+ * @see <a href="http://json-schema.org/">http://json-schema.org/</a>
+ */
+ public JsonRowSerializationSchema(String jsonSchema) {
+ this(JsonSchemaConverter.convert(jsonSchema));
+ }
+
+ @Override
+ public byte[] serialize(Row row) {
+ if (node == null) {
+ node = mapper.createObjectNode();
+ }
+
+ try {
+ convertRow(node, (RowTypeInfo) typeInfo, row);
+ return mapper.writeValueAsBytes(node);
+ } catch (Throwable t) {
+ throw new RuntimeException("Could not serialize row '" + row + "'. " +
+ "Make sure that the schema matches the input.", t);
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ private ObjectNode convertRow(ObjectNode reuse, RowTypeInfo info, Row row) {
+ if (reuse == null) {
+ reuse = mapper.createObjectNode();
+ }
+ final String[] fieldNames = info.getFieldNames();
+ final TypeInformation<?>[] fieldTypes = info.getFieldTypes();
+
+ // validate the row
+ if (row.getArity() != fieldNames.length) {
+ throw new IllegalStateException(String.format(
+ "Number of elements in the row '%s' is different from number of field names: %d", row, fieldNames.length));
+ }
+
+ for (int i = 0; i < fieldNames.length; i++) {
+ final String name = fieldNames[i];
+
+ final JsonNode fieldConverted = convert(reuse, reuse.get(name), fieldTypes[i], row.getField(i));
+ reuse.set(name, fieldConverted);
+ }
+
+ return reuse;
+ }
+
+ 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) {
+ // convert decimal if necessary
+ if (object instanceof BigDecimal) {
+ return container.numberNode((BigDecimal) object);
+ }
+ return container.numberNode(BigDecimal.valueOf(((Number) object).doubleValue()));
+ } else if (info == Types.BIG_INT) {
+ // convert integer if necessary
+ 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;
+ // strip milliseconds if possible
+ 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 {
+ // for types that were specified without JSON schema
+ // e.g. POJOs
+ try {
+ return mapper.valueToTree(object);
+ } catch (IllegalArgumentException e) {
+ throw new IllegalStateException("Unsupported type information '" + info + "' for object: " + object, e);
+ }
+ }
+ }
+
+ private ArrayNode convertObjectArray(ArrayNode reuse, TypeInformation<?> info, Object[] array) {
+ if (reuse == null) {
+ reuse = mapper.createArrayNode();
+ } else {
+ reuse.removeAll();
+ }
+
+ for (Object object : array) {
+ reuse.add(convert(reuse, null, info, object));
+ }
+ return reuse;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonSchemaConverter.java
----------------------------------------------------------------------
diff --git a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonSchemaConverter.java b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonSchemaConverter.java
new file mode 100644
index 0000000..58495b0
--- /dev/null
+++ b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonSchemaConverter.java
@@ -0,0 +1,361 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.formats.json;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonParser;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+/**
+ * Converts a JSON schema into Flink's type information. It uses {@link Row} for representing
+ * objects and tuple arrays.
+ *
+ * <p>Note: This converter implements just a subset of the JSON schema specification.
+ * Union types (as well as "allOf", "anyOf", "not") are not supported yet. Simple
+ * references that link to a common definition in the document are supported. "oneOf" and
+ * arrays of types are only supported for specifying nullability.
+ *
+ * <p>This converter has been developed for JSON Schema draft-07 but also includes keywords of
+ * older drafts to be as compatible as possible.
+ */
+@SuppressWarnings("OptionalIsPresent")
+public final class JsonSchemaConverter {
+
+ private JsonSchemaConverter() {
+ // private
+ }
+
+ // see https://spacetelescope.github.io/understanding-json-schema/UnderstandingJSONSchema.pdf
+ private static final String PROPERTIES = "properties";
+ private static final String ADDITIONAL_PROPERTIES = "additionalProperties";
+ private static final String TYPE = "type";
+ private static final String FORMAT = "format";
+ private static final String CONTENT_ENCODING = "contentEncoding";
+ private static final String ITEMS = "items";
+ private static final String ADDITIONAL_ITEMS = "additionalItems";
+ private static final String REF = "$ref";
+ private static final String ALL_OF = "allOf";
+ private static final String ANY_OF = "anyOf";
+ private static final String NOT = "not";
+ private static final String ONE_OF = "oneOf";
+
+ // from https://tools.ietf.org/html/draft-zyp-json-schema-03#page-14
+ private static final String DISALLOW = "disallow";
+ private static final String EXTENDS = "extends";
+
+ private static final String TYPE_NULL = "null";
+ private static final String TYPE_BOOLEAN = "boolean";
+ private static final String TYPE_OBJECT = "object";
+ private static final String TYPE_ARRAY = "array";
+ private static final String TYPE_NUMBER = "number";
+ private static final String TYPE_INTEGER = "integer";
+ private static final String TYPE_STRING = "string";
+
+ private static final String FORMAT_DATE = "date";
+ private static final String FORMAT_TIME = "time";
+ private static final String FORMAT_DATE_TIME = "date-time";
+
+ private static final String CONTENT_ENCODING_BASE64 = "base64";
+
+ /**
+ * Converts a JSON schema into Flink's type information. Throws an exception if the schema
+ * cannot converted because of loss of precision or too flexible schema.
+ *
+ * <p>The converter can resolve simple schema references to solve those cases where entities
+ * are defined at the beginning and then used throughout a document.
+ */
+ @SuppressWarnings("unchecked")
+ public static <T> TypeInformation<T> convert(String jsonSchema) {
+ Preconditions.checkNotNull(jsonSchema, "JSON schema");
+ final ObjectMapper mapper = new ObjectMapper();
+ mapper.getFactory()
+ .enable(JsonParser.Feature.ALLOW_COMMENTS)
+ .enable(JsonParser.Feature.ALLOW_UNQUOTED_FIELD_NAMES)
+ .enable(JsonParser.Feature.ALLOW_SINGLE_QUOTES);
+ final JsonNode node;
+ try {
+ node = mapper.readTree(jsonSchema);
+ } catch (IOException e) {
+ throw new IllegalArgumentException(
+ "Invalid JSON schema.", e);
+ }
+ return (TypeInformation<T>) convertType("<root>", node, node);
+ }
+
+ private static TypeInformation<?> convertType(String location, JsonNode node, JsonNode root) {
+ // we use a set here to unify types (e.g. types that just add metadata such as 'multipleOf')
+ final Set<TypeInformation<?>> typeSet = new HashSet<>();
+
+ // search for ref
+ final Optional<JsonNode> ref;
+ if (node.has(REF) && node.get(REF).isTextual()) {
+ // try a simple ref resolver to solve those cases where entities are defined at
+ // the beginning and then used throughout a document
+ ref = Optional.of(resolveReference(node.get(REF).asText(), node, root));
+ } else {
+ ref = Optional.empty();
+ }
+
+ // use TYPE of this node
+ if (node.has(TYPE)) {
+ final JsonNode typeNode = node.get(TYPE);
+
+ List<String> types = new ArrayList<>();
+ // array of types
+ if (typeNode.isArray()) {
+ final Iterator<JsonNode> elements = typeNode.elements();
+ while (elements.hasNext()) {
+ types.add(elements.next().asText());
+ }
+ }
+ // single type
+ else if (typeNode.isTextual()) {
+ types.add(typeNode.asText());
+ }
+
+ for (String type : types) {
+ // set field type
+ switch (type) {
+ case TYPE_NULL:
+ typeSet.add(Types.VOID);
+ break;
+ case TYPE_BOOLEAN:
+ typeSet.add(Types.BOOLEAN);
+ break;
+ case TYPE_STRING:
+ if (node.has(FORMAT)) {
+ typeSet.add(convertStringFormat(location, node.get(FORMAT)));
+ } else if (node.has(CONTENT_ENCODING)) {
+ typeSet.add(convertStringEncoding(location, node.get(CONTENT_ENCODING)));
+ } else {
+ typeSet.add(Types.STRING);
+ }
+ break;
+ case TYPE_NUMBER:
+ typeSet.add(Types.BIG_DEC);
+ break;
+ case TYPE_INTEGER:
+ typeSet.add(Types.BIG_INT);
+ break;
+ case TYPE_OBJECT:
+ typeSet.add(convertObject(location, node, root));
+ break;
+ case TYPE_ARRAY:
+ typeSet.add(convertArray(location, node, root));
+ break;
+ default:
+ throw new IllegalArgumentException(
+ "Unsupported type '" + node.get(TYPE).asText() + "' in node: " + location);
+ }
+ }
+ }
+ // use TYPE of reference as fallback
+ else if (ref.isPresent() && ref.get().has(TYPE)) {
+ typeSet.add(convertType(node.get(REF).asText(), ref.get(), root));
+ }
+
+ // simple interpretation of ONE_OF for supporting "object or null"
+ if (node.has(ONE_OF) && node.get(ONE_OF).isArray()) {
+ final TypeInformation<?>[] types = convertTypes(location + '/' + ONE_OF, node.get(ONE_OF), root);
+ typeSet.addAll(Arrays.asList(types));
+ }
+ // use ONE_OF of reference as fallback
+ else if (ref.isPresent() && ref.get().has(ONE_OF) && ref.get().get(ONE_OF).isArray()) {
+ final TypeInformation<?>[] types = convertTypes(node.get(REF).asText() + '/' + ONE_OF, ref.get().get(ONE_OF), root);
+ typeSet.addAll(Arrays.asList(types));
+ }
+
+ // validate no union types or extending
+ if (node.has(ALL_OF) || node.has(ANY_OF) || node.has(NOT) || node.has(EXTENDS) || node.has(DISALLOW)) {
+ throw new IllegalArgumentException(
+ "Union types are such as '" + ALL_OF + "', '" + ANY_OF + "' etc. " +
+ "and extending are not supported yet.");
+ }
+
+ // only a type (with null) is supported yet
+ final List<TypeInformation<?>> types = new ArrayList<>(typeSet);
+ if (types.size() == 0) {
+ throw new IllegalArgumentException("No type could be found in node:" + location);
+ } else if (types.size() > 2 || (types.size() == 2 && !types.contains(Types.VOID))) {
+ throw new IllegalArgumentException(
+ "Union types with more than just a null type are not supported yet.");
+ }
+
+ // return the first non-void type or void
+ if (types.size() == 2 && types.get(0) == Types.VOID) {
+ return types.get(1);
+ } else {
+ return types.get(0);
+ }
+ }
+
+ private static TypeInformation<Row> convertObject(String location, JsonNode node, JsonNode root) {
+ // validate properties
+ if (!node.has(PROPERTIES)) {
+ return Types.ROW();
+ }
+ if (!node.isObject()) {
+ throw new IllegalArgumentException(
+ "Invalid '" + PROPERTIES + "' property for object type in node: " + location);
+ }
+ final JsonNode props = node.get(PROPERTIES);
+ final String[] names = new String[props.size()];
+ final TypeInformation<?>[] types = new TypeInformation[props.size()];
+
+ final Iterator<Map.Entry<String, JsonNode>> fieldIter = props.fields();
+ int i = 0;
+ while (fieldIter.hasNext()) {
+ final Map.Entry<String, JsonNode> subNode = fieldIter.next();
+
+ // set field name
+ names[i] = subNode.getKey();
+
+ // set type
+ types[i] = convertType(location + '/' + subNode.getKey(), subNode.getValue(), root);
+
+ i++;
+ }
+
+ // validate that object does not contain additional properties
+ if (node.has(ADDITIONAL_PROPERTIES) && node.get(ADDITIONAL_PROPERTIES).isBoolean() &&
+ node.get(ADDITIONAL_PROPERTIES).asBoolean()) {
+ throw new IllegalArgumentException(
+ "An object must not allow additional properties in node: " + location);
+ }
+
+ return Types.ROW_NAMED(names, types);
+ }
+
+ private static TypeInformation<?> convertArray(String location, JsonNode node, JsonNode root) {
+ // validate items
+ if (!node.has(ITEMS)) {
+ throw new IllegalArgumentException(
+ "Arrays must specify an '" + ITEMS + "' property in node: " + location);
+ }
+ final JsonNode items = node.get(ITEMS);
+
+ // list (translated to object array)
+ if (items.isObject()) {
+ final TypeInformation<?> elementType = convertType(
+ location + '/' + ITEMS,
+ items,
+ root);
+ // result type might either be ObjectArrayTypeInfo or BasicArrayTypeInfo for Strings
+ return Types.OBJECT_ARRAY(elementType);
+ }
+ // tuple (translated to row)
+ else if (items.isArray()) {
+ final TypeInformation<?>[] types = convertTypes(location + '/' + ITEMS, items, root);
+
+ // validate that array does not contain additional items
+ if (node.has(ADDITIONAL_ITEMS) && node.get(ADDITIONAL_ITEMS).isBoolean() &&
+ node.get(ADDITIONAL_ITEMS).asBoolean()) {
+ throw new IllegalArgumentException(
+ "An array tuple must not allow additional items in node: " + location);
+ }
+
+ return Types.ROW(types);
+ }
+ throw new IllegalArgumentException(
+ "Invalid type for '" + ITEMS + "' property in node: " + location);
+ }
+
+ private static TypeInformation<?> convertStringFormat(String location, JsonNode node) {
+ if (!node.isTextual()) {
+ throw new IllegalArgumentException("Invalid '" + FORMAT + "' property in node: " + location);
+ }
+
+ switch (node.asText()) {
+ case FORMAT_DATE:
+ return Types.SQL_DATE;
+ case FORMAT_TIME:
+ return Types.SQL_TIME;
+ case FORMAT_DATE_TIME:
+ return Types.SQL_TIMESTAMP;
+ default:
+ return Types.STRING; // unlikely that we will support other formats in the future
+ }
+ }
+
+ private static TypeInformation<?> convertStringEncoding(String location, JsonNode node) {
+ if (!node.isTextual()) {
+ throw new IllegalArgumentException("Invalid '" + CONTENT_ENCODING + "' property in node: " + location);
+ }
+
+ // "If the instance value is a string, this property defines that the string SHOULD
+ // be interpreted as binary data and decoded using the encoding named by this property."
+
+ switch (node.asText()) {
+ case CONTENT_ENCODING_BASE64:
+ return Types.PRIMITIVE_ARRAY(Types.BYTE);
+ default:
+ // we fail hard here:
+ // this gives us the chance to support more encodings in the future without problems
+ // of backwards compatibility
+ throw new IllegalArgumentException("Invalid encoding '" + node.asText() + "' in node: " + location);
+ }
+ }
+
+ private static JsonNode resolveReference(String ref, JsonNode origin, JsonNode root) {
+ if (!ref.startsWith("#")) {
+ throw new IllegalArgumentException("Only JSON schemes with simple references " +
+ "(one indirection in the same document) are supported yet. But was: " + ref);
+ }
+ final String path = ref.substring(1);
+ final JsonNode foundNode = root.at(path);
+ if (foundNode.isMissingNode()) {
+ throw new IllegalArgumentException("Could not find reference: " + ref);
+ }
+ // prevent obvious cyclic references
+ if (foundNode == origin) {
+ throw new IllegalArgumentException("Cyclic references are not supported:" + ref);
+ }
+ return foundNode;
+ }
+
+ private static TypeInformation<?>[] convertTypes(String location, JsonNode arrayNode, JsonNode root) {
+ final TypeInformation<?>[] types = new TypeInformation[arrayNode.size()];
+ final Iterator<JsonNode> elements = arrayNode.elements();
+ int i = 0;
+ while (elements.hasNext()) {
+ final TypeInformation<?> elementType = convertType(
+ location + '[' + i + ']',
+ elements.next(),
+ root);
+ types[i] = elementType;
+ i += 1;
+ }
+ return types;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/6c09c057/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonNodeDeserializationSchemaTest.java
----------------------------------------------------------------------
diff --git a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonNodeDeserializationSchemaTest.java b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonNodeDeserializationSchemaTest.java
new file mode 100644
index 0000000..1462a26
--- /dev/null
+++ b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonNodeDeserializationSchemaTest.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.formats.json;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
+
+import org.junit.Test;
+
+import java.io.IOException;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Tests for the {@link JsonNodeDeserializationSchema}.
+ */
+public class JsonNodeDeserializationSchemaTest {
+
+ @Test
+ public void testDeserialize() throws IOException {
+ ObjectMapper mapper = new ObjectMapper();
+ ObjectNode initialValue = mapper.createObjectNode();
+ initialValue.put("key", 4).put("value", "world");
+ byte[] serializedValue = mapper.writeValueAsBytes(initialValue);
+
+ JsonNodeDeserializationSchema schema = new JsonNodeDeserializationSchema();
+ ObjectNode deserializedValue = schema.deserialize(serializedValue);
+
+ assertEquals(4, deserializedValue.get("key").asInt());
+ assertEquals("world", deserializedValue.get("value").asText());
+ }
+}