flink mongodb sink

Flink 提供了一个 MongoDB 的 Sink，可以将数据流写入 MongoDB 数据库。在使用 MongoDB Sink 之前，需要先添加 Flink 的 MongoDB Connector 依赖。可以使用以下 Maven 依赖：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-mongodb</artifactId>
    <version>${flink.version}</version>
</dependency>

在代码中可以通过以下方式创建一个 MongoDB Sink：

MongoDBSink sink = MongoDBSink.<MyType>builder()
    .withUri("mongodb://localhost:27017")
    .withDatabase("mydb")
    .withCollection("mycollection")
    .withMapper(new MyTypeMapper())
    .build();

其中，`MyType` 是要写入到 MongoDB 的数据类型，`MyTypeMapper` 是将数据类型转换为 MongoDB 中的文档格式的映射器。在创建 Sink 后，可以将其添加到 Flink 的 DataStream 中，例如：

DataStream<MyType> stream = ...
stream.addSink(sink);

这样，数据流就会被写入到指定的 MongoDB 数据库和集合中。