我想设置Flink,以便将数据流从Apache Kafka转换并重定向到MongoDB.出于测试目的,我建立在flink-streaming-connectors.kafka示例(https://github.com/apache/flink)之上.
Faf正在为Kafka流提供正确的红色,我可以映射它们等等,但是当我想将每个收到的和转换后的消息保存到MongoDB时会出现问题.我发现的关于MongoDB集成的唯一例子是来自github的flink-mongodb-test.不幸的是,它使用静态数据源(数据库),而不是数据流.
我相信MongoDB应该有一些DataStream.addSink实现,但显然没有.
实现它的最佳方法是什么？我是否需要编写自定义接收器功能或者我可能缺少某些东西？也许它应该以不同的方式完成？
我没有任何解决方案,所以任何建议将不胜感激.
下面是一个例子,我正在获得什么作为输入以及我需要存储为输出.
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的消息流缓冲和一些基本的解析.