feat: initfeature

3 files changed, 87 insertions, 0 deletions

diff --git a/src/main/java/com/chaoc/flink/serialization/JsonTest.java b/src/main/java/com/chaoc/flink/serialization/JsonTest.java
new file mode 100644
index 0000000..f00fabe
--- /dev/null
+++ b/src/main/java/com/chaoc/flink/serialization/JsonTest.java
@@ -0,0 +1,42 @@
+package com.chaoc.flink.serialization;
+
+import org.apache.flink.formats.json.JsonNodeDeserializationSchema;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
+import com.chaoc.flink.serialization.formats.JsonNodeSerializationSchema;
+
+import java.util.Properties;
+
+public class JsonTest {
+
+    public static void main(String[] args) throws Exception {
+
+       final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+        final Properties srcProps = new Properties();
+        srcProps.setProperty("bootstrap.servers", "192.168.41.32:9092");
+        srcProps.setProperty("group.id", "jackson-serialization");
+
+        final FlinkKafkaConsumer<ObjectNode> consumer = new FlinkKafkaConsumer<>(
+                "SESSION-RECORD-JSON",
+                new JsonNodeDeserializationSchema(),
+                srcProps);
+
+        final Properties dstProps = new Properties();
+        dstProps.setProperty("bootstrap.servers", "192.168.41.32:9092");
+
+        final FlinkKafkaProducer<ObjectNode> producer = new FlinkKafkaProducer<>(
+                "SESSION-RECORD-JSON-COMPLETED",
+                new JsonNodeSerializationSchema(),
+                dstProps);
+
+        env.addSource(consumer)
+                .name("SessionRecordJacksonReader")
+                .addSink(producer)
+                .name("SessionRecordJacksonWriter");
+
+        env.execute("Jackson Serialization Test");
+    }
+}
diff --git a/src/main/java/com/chaoc/flink/serialization/formats/JsonNodeSerializationSchema.java b/src/main/java/com/chaoc/flink/serialization/formats/JsonNodeSerializationSchema.java
new file mode 100644
index 0000000..0de2f56
--- /dev/null
+++ b/src/main/java/com/chaoc/flink/serialization/formats/JsonNodeSerializationSchema.java
@@ -0,0 +1,20 @@
+package com.chaoc.flink.serialization.formats;
+
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
+
+public class JsonNodeSerializationSchema implements SerializationSchema<ObjectNode> {
+
+    private final ObjectMapper mapper = new ObjectMapper();
+
+    @Override
+    public byte[] serialize(ObjectNode element) {
+        try {
+            return mapper.writeValueAsBytes(element);
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException(e);
+        }
+    }
+}
diff --git a/src/main/resources/log4j2.properties b/src/main/resources/log4j2.properties
new file mode 100644
index 0000000..e7c4ef9
--- /dev/null
+++ b/src/main/resources/log4j2.properties
@@ -0,0 +1,25 @@
+################################################################################
+#  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.
+################################################################################
+
+rootLogger.level = WARN
+rootLogger.appenderRef.console.ref = ConsoleAppender
+
+appender.console.name = ConsoleAppender
+appender.console.type = CONSOLE
+appender.console.layout.type = PatternLayout
+appender.console.layout.pattern = %d{HH:mm:ss,SSS} %-5p %-60c %x - %m%n