search

Johnwickdev / Hightable

0 stars 0 forks source link

Class1 #97

Open Johnwickdev opened 1 month ago

Johnwickdev commented 1 month ago

import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.connector.mongodb.source.MongoSource; import org.apache.flink.connector.mongodb.MongoSourceOptions; import org.apache.flink.api.common.serialization.SimpleStringSchema;

public class MongoDBSourceFlink {

public static DataStream<String> createMongoDBSource(StreamExecutionEnvironment env) {
    MongoSource<String> mongoSource = MongoSource.builder()
        .setUri("mongodb://yourMongoUri") // MongoDB URI
        .setDatabase("yourDatabase")
        .setCollection("yourCollection")
        .setDeserializationSchema(new SimpleStringSchema()) // Schema to deserialize MongoDB JSON data
        .build();

    return env.fromSource(mongoSource, MongoSourceOptions.noBoundedness(), "MongoDBSource");
}

}