[
https://issues.apache.org/jira/browse/FLINK-3874?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15420261#comment-15420261
]
ASF GitHub Bot commented on FLINK-3874:
---------------------------------------
Github user mushketyk commented on a diff in the pull request:
https://github.com/apache/flink/pull/2244#discussion_r74698003
--- Diff:
flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/util/serialization/JsonRowSerializationSchema.java
---
@@ -0,0 +1,69 @@
+/*
+ * 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.util.serialization;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.flink.api.table.Row;
+import org.apache.flink.util.Preconditions;
+
+
+/**
+ * Serialization schema that serializes an object into a JSON bytes.
+ *
+ * <p>Serializes the input {@link Row} object into a JSON string and
+ * converts it into <code>byte[]</code>.
+ *
+ * <p>Result <bode>byte[]</bode> messages can be deserialized using
+ * {@link JsonRowDeserializationSchema}.
+ */
+public class JsonRowSerializationSchema implements
SerializationSchema<Row> {
+ /** 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();
+
+ /**
+ * Creates a JSON serialization schema for the given fields and types.
+ *
+ * @param fieldNames Names of JSON fields to parse.
+ */
+ public JsonRowSerializationSchema(String[] fieldNames) {
+ this.fieldNames = Preconditions.checkNotNull(fieldNames);
+ }
+
+ @Override
+ public byte[] serialize(Row row) {
+ if (row.productArity() != 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.productArity(); i++) {
+ JsonNode node =
mapper.valueToTree(row.productElement(i));
+ objectNode.set(fieldNames[i], node);
+ }
+
+ try {
+ return mapper.writeValueAsBytes(objectNode);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to serialize row",
e);
--- End diff --
I think it would be better, but I implement **SerializationSchema**
interface here and it declare that it throws any exceptions. I could change the
interface, but it is marked as Public so I am not sure if this is a good idea.
> Add a Kafka TableSink with JSON serialization
> ---------------------------------------------
>
> Key: FLINK-3874
> URL: https://issues.apache.org/jira/browse/FLINK-3874
> Project: Flink
> Issue Type: New Feature
> Components: Table API & SQL
> Reporter: Fabian Hueske
> Assignee: Ivan Mushketyk
> Priority: Minor
>
> Add a TableSink that writes JSON serialized data to Kafka.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
