I would recommend checking out the Flink RabbitMQ Source for examples:

For your case, you should extend the `RichSourceFunction` which provides 
additional access to override the `open()` life cycle method.
In that method, you instantiate your MongoDB client connection and  fetch the 
cursor. In the `run()` method, you should essentially have a while loop that 
polls the MongoDB cursor and emits the fetched documents using the 

If your also looking to implement a MongoDB source that works with Flink’s 
checkpointing for exactly-once, be sure to check out:

On February 16, 2017 at 5:53:03 PM, Pedro Monteiro (pedro.mlmonte...@gmail.com) 

Dear Tzu-Li,

Thank you so much for your prompt response.

Lets assume I have a variable, in Java, env which is my 
StreamExecutionEnvironment. When I go ahead and attempt to do:
​It requests an implementation of a Source Function interface: ​
env.addSource(new SourceFunction<Document>() {
            ​​ @Override
            public void run(SourceFunction.SourceContext<Document> ctx) throws 
Exception {
                 ​// TO DO​

            public void cancel() {
                ​// TO DO​
​And this is where I'm somehow stuck. I do not understand how should I access 
my MongoDB's cursor in any of this methods (I suppose the most adequate would 
be the "run" method) in a way it would allow me to return a new MongoDB 
document as it arrived to the database from another source.

Once again, thank you so much for your help.

I will wait to hear from you!​


Pedro Lima Monteiro

On 16 February 2017 at 09:43, Tzu-Li (Gordon) Tai <tzuli...@apache.org> wrote:
Hi Pedro!

This is definitely possible, by simply writing a Flink `SourceFunction` that 
uses MongoDB clients to fetch the data.
It should be straightforward and works well with MongoDB’s cursor APIs.

Could you explain a bit which part in particular you were stuck with?


On February 16, 2017 at 5:31:39 PM, Pedro Monteiro (pedro.mlmonte...@gmail.com) 

Good morning,

I am trying to get data from MongoDB to be analysed in Flink.
I would like to know if it is possible to stream data from MongoDB into
Flink. I have looked into Flink's source function to add in the addSource
method of the StreamExecutionEnvironment but I had no luck.
Can anyone help me out?

Pedro Lima Monteiro

Reply via email to