Apache Kafka - sgml/signature GitHub Wiki

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-java</artifactId>
    <version>1.15.0</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-java_2.12</artifactId>
    <version>1.15.0</version>
</dependency>
<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-core</artifactId>
    <version>2.13.0</version>
</dependency>
<dependency>
    <groupId>com.github.fge</groupId>
    <artifactId>jackson-coreutils</artifactId>
    <version>1.9</version>
</dependency>

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import com.fasterxml.jackson.core.JsonPointer;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;

public class JsonPathFinder {
    public static void main(String[] args) throws Exception {
        // Set up the streaming execution environment
        var env = StreamExecutionEnvironment.getExecutionEnvironment();

        // Create a sample JSON input stream
        var jsonInputStream = env.fromElements(
                """
                {"foo": "bar"}
                """,
                """
                {"baz": "qux"}
                """
        );

        // Define a FlatMapFunction to find the JSON Pointer path "foo"
        var results = jsonInputStream.flatMap((FlatMapFunction<String, Tuple2<String, Boolean>>) (value, out) -> {
            var mapper = new ObjectMapper();
            var rootNode = mapper.readTree(value);
            var pointer = JsonPointer.compile("/foo");

            var pathExists = !rootNode.at(pointer).isMissingNode();
            out.collect(new Tuple2<>(value, pathExists));
        });

        // Print the results
        results.print();

        // Execute the Flink job
        env.execute("Json Path Finder");
    }
}