Main

[Johnwickdev]

import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.api.common.serialization.DeserializationSchema; import org.apache.flink.api.common.serialization.SimpleStringSchema; import org.apache.flink.connector.mongodb.source.MongoSource; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.bson.Document;

import java.io.IOException;

public class FlinkMongoReadTest {

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

    // Configure the MongoDB source with a custom DeserializationSchema
    MongoSource<String> mongoSource = MongoSource.<String>builder()
            .setUri("your-mongodb-uri")  // Replace with your MongoDB URI
            .setDatabase("your-database-name")  // Replace with your database name
            .setCollection("your-collection-name")  // Replace with your collection name
            .setDeserializationSchema(new DocumentStringDeserializationSchema())  // Custom DeserializationSchema
            .build();

    // Create a data stream from the MongoDB source
    DataStream<String> mongoStream = env.fromSource(
            mongoSource,
            WatermarkStrategy.noWatermarks(),
            "MongoDBSource"
    );

    // Print the documents read from MongoDB
    mongoStream.print();

    // Execute the Flink job
    env.execute("Flink MongoDB Read Test");
}

// Custom DeserializationSchema to convert MongoDB Document to String
public static class DocumentStringDeserializationSchema implements DeserializationSchema<String> {

    @Override
    public String deserialize(byte[] message) throws IOException {
        Document doc = Document.parse(new String(message));
        return doc.toJson();
    }

    @Override
    public boolean isEndOfStream(String nextElement) {
        return false;
    }

    @Override
    public org.apache.flink.api.common.typeinfo.TypeInformation<String> getProducedType() {
        return org.apache.flink.api.common.typeinfo.TypeInformation.of(String.class);
    }
}

}