我想设置Flink,以便将来自Apache Kafka的数据流转换并重定向到MongoDB。为了测试目的,我正在基于flink-streaming-connectors.kafka示例进行构建(https://github.com/apache/flink)。
Kafka流被Flink正确地读取,我可以对它们进行映射等操作,但是当我想将每个接收到和转换后的消息保存到MongoDB时,问题就出现了。我在github上找到了关于MongoDB集成的唯一示例是flink-mongodb-test。不幸的是,它使用静态数据源(数据库),而不是数据流。
我相信应该有一些DataStream.addSink实现用于MongoDB,但显然没有。
如何才能最好地实现它?我需要编写自定义sink函数吗,还是我漏掉了什么?也许应该用不同的方法来完成?
我没有绑定任何解决方案,所以欢迎任何建议。
下面是一个示例,说明我输入的内容以及我需要存储的输出内容。
Apache Kafka Broker <-------------- "AAABBBCCCDDD" (String)
Apache Kafka Broker --------------> Flink: DataStream<String>
Flink: DataStream.map({
return ("AAABBBCCCDDD").convertTo("A: AAA; B: BBB; C: CCC; D: DDD")
})
.rebalance()
.addSink(MongoDBSinkFunction); // store the row in MongoDB collection
正如你在这个例子中看到的,我主要使用Flink来缓冲Kafka的消息流和进行一些基本的解析。