[FLINK-8630] [formats] Add proper support for JSON formats

This closes #5491.
apache · Feb 15, 2018 · 6c09c05 · 6c09c05
1 parent 942649e
commit 6c09c05
Show file tree

Hide file tree

Showing 33 changed files with 1,751 additions and 437 deletions.
diff --git 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>
@@ -100,6 +108,14 @@ under the License.
 			<type>test-jar</type>
 		</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.flink</groupId>
 			<artifactId>flink-streaming-java_${scala.binary.version}</artifactId>

diff --git a/.../src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSinkTest.java b/.../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;

diff --git a/...rc/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSourceTest.java b/...rc/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;
 
 /**

diff --git 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>
@@ -109,6 +117,14 @@ under the License.
 			<type>test-jar</type>
 		</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.flink</groupId>
 			<artifactId>flink-streaming-java_${scala.binary.version}</artifactId>

diff --git a/...rc/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011JsonTableSourceTest.java b/...rc/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;
 
 /**

diff --git a/flink-connectors/flink-connector-kafka-0.8/pom.xml b/flink-connectors/flink-connector-kafka-0.8/pom.xml
@@ -81,6 +81,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>
+
 		<dependency>
 			<groupId>org.apache.kafka</groupId>
 			<artifactId>kafka_${scala.binary.version}</artifactId>
@@ -143,6 +151,14 @@ under the License.
 			<type>test-jar</type>
 		</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>

diff --git a/...8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSinkTest.java b/...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;

diff --git a/...src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSourceTest.java b/...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;
 
 /**

diff --git a/flink-connectors/flink-connector-kafka-0.9/pom.xml b/flink-connectors/flink-connector-kafka-0.9/pom.xml
@@ -80,6 +80,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>
+
 		<dependency>
 			<groupId>org.apache.kafka</groupId>
 			<artifactId>kafka-clients</artifactId>
@@ -96,6 +104,14 @@ under the License.
 			<type>test-jar</type>
 		</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.flink</groupId>
 			<artifactId>flink-streaming-java_${scala.binary.version}</artifactId>

diff --git a/...9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09JsonTableSinkTest.java b/...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;

diff --git a/...src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09JsonTableSourceTest.java b/...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;
 
 /**

diff --git a/flink-connectors/flink-connector-kafka-base/pom.xml b/flink-connectors/flink-connector-kafka-base/pom.xml
@@ -74,6 +74,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>
+
 		<dependency>
 			<groupId>org.apache.kafka</groupId>
 			<artifactId>kafka_${scala.binary.version}</artifactId>
@@ -182,6 +190,14 @@ under the License.
 			<scope>test</scope>
 		</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>

diff --git a/...ka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSink.java b/...ka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSink.java
@@ -21,8 +21,8 @@
 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;

diff --git a/...-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSource.java b/...-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSource.java
@@ -21,7 +21,7 @@
 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;

diff --git a/...rc/main/java/org/apache/flink/streaming/util/serialization/JSONDeserializationSchema.java b/...rc/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(&lt;name>).as(&lt;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
 }
diff --git a/...main/java/org/apache/flink/streaming/util/serialization/JsonRowDeserializationSchema.java b/...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 @@
  * 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);
 	}
-
 }