How MapFunction gets executed?

[Yan Chou Chen]

A quick question. When running a stream job that executes
DataStream.map(MapFunction) , after data is read from Kafka, does each
MapFunction is created per item or based on parallelism?

For instance, for the following code snippet

val env = StreamExecutionEnvironment.getExeutionEnvironment
val stream = env.addSource(FlinkKafkaConsumer09(...))
stream.map(new RichMapFunction[String, Unit] {

    // my AsyncHttpClient instance

    override def open(params: Configuration) { /* create my
AsyncHttpClient instance, etc. */ }

    override def close() { /* close my AsyncHttpClient instance*/ }

    override def map(record: String) {
        // my code
    }
})

Is RichMapFunction created for each record (as String in the above
example)? Or say the program set parallelism to 4 so 4 RichMapFunction
instances are created first, then data read from Kafka consumer is
divided into 4 partitions (or something similar), and then map(record:
String) is called within something like while loop? Or what is the
actual flow? Or source code I can start from (I trace through
StreamExecutionEnvironment/ addSource/ DataStream/ transform/
addOperator etc., but I then get lost in source code)?

Basically my problem is I have an AsyncHttpClient instance opened
within open() function and close in close function according to the
RichMapFunction doc. However, an issue is that in some cases my
AsyncHttpClient instance is not executed which displays warning like

AsyncHttpClient.close() hasn't been invoked, which may produce file
descriptor leaks

Therefore I would like to know the life cycle so that I can close
resource appropriately.

Thanks

[Till Rohrmann]

Hi Yan Chou Chen,

Flink does not instantiate for each record a mapper. Instead, it will create as many mappers as you've defined with the parallelism. Each of these mappers is deployed to a slot on a TaskManager. When it is deployed and before it receives records, the open method is called once. Then incoming records are processed as they arrive at the operator. Once the operator has finished processing (in the streaming case, this means that the user has stopped or cancelled the job) it will call the close method. The close method should also be called if your job fails. Therefore, I cannot explain why some of your resources don't get closed. Could you check whether the logs contains something suspicious.

Cheers,

Till

On Thu, Jun 16, 2016 at 5:07 PM, Yan Chou Chen <[hidden email]> wrote:

A quick question. When running a stream job that executes
DataStream.map(MapFunction) , after data is read from Kafka, does each
MapFunction is created per item or based on parallelism?

For instance, for the following code snippet

val env = StreamExecutionEnvironment.getExeutionEnvironment
val stream = env.addSource(FlinkKafkaConsumer09(...))
stream.map(new RichMapFunction[String, Unit] {

    // my AsyncHttpClient instance

    override def open(params: Configuration) { /* create my
AsyncHttpClient instance, etc. */ }

    override def close() { /* close my AsyncHttpClient instance*/ }

    override def map(record: String) {
        // my code
    }
})

Is RichMapFunction created for each record (as String in the above
example)? Or say the program set parallelism to 4 so 4 RichMapFunction
instances are created first, then data read from Kafka consumer is
divided into 4 partitions (or something similar), and then map(record:
String) is called within something like while loop? Or what is the
actual flow? Or source code I can start from (I trace through
StreamExecutionEnvironment/ addSource/ DataStream/ transform/
addOperator etc., but I then get lost in source code)?

Basically my problem is I have an AsyncHttpClient instance opened
within open() function and close in close function according to the
RichMapFunction doc. However, an issue is that in some cases my
AsyncHttpClient instance is not executed which displays warning like

AsyncHttpClient.close() hasn't been invoked, which may produce file
descriptor leaks

Therefore I would like to know the life cycle so that I can close
resource appropriately.

Thanks

[Yan Chou Chen]

Thanks for clarifying that helps me identify the root cause. The
problem comes from my code which is not related to Flink. Now the
problem is solved. Thank you again for the advice!

On 16 June 2016 at 23:49, Till Rohrmann <[hidden email]> wrote:

> Hi Yan Chou Chen,
>
> Flink does not instantiate for each record a mapper. Instead, it will create
> as many mappers as you've defined with the parallelism. Each of these
> mappers is deployed to a slot on a TaskManager. When it is deployed and
> before it receives records, the open method is called once. Then incoming
> records are processed as they arrive at the operator. Once the operator has
> finished processing (in the streaming case, this means that the user has
> stopped or cancelled the job) it will call the close method. The close
> method should also be called if your job fails. Therefore, I cannot explain
> why some of your resources don't get closed. Could you check whether the
> logs contains something suspicious.
>
> Cheers,
> Till
>
> On Thu, Jun 16, 2016 at 5:07 PM, Yan Chou Chen <[hidden email]> wrote:
>>
>> A quick question. When running a stream job that executes
>> DataStream.map(MapFunction) , after data is read from Kafka, does each
>> MapFunction is created per item or based on parallelism?
>>
>> For instance, for the following code snippet
>>
>> val env = StreamExecutionEnvironment.getExeutionEnvironment
>> val stream = env.addSource(FlinkKafkaConsumer09(...))
>> stream.map(new RichMapFunction[String, Unit] {
>>
>>     // my AsyncHttpClient instance
>>
>>     override def open(params: Configuration) { /* create my
>> AsyncHttpClient instance, etc. */ }
>>
>>     override def close() { /* close my AsyncHttpClient instance*/ }
>>
>>     override def map(record: String) {
>>         // my code
>>     }
>> })
>>
>> Is RichMapFunction created for each record (as String in the above
>> example)? Or say the program set parallelism to 4 so 4 RichMapFunction
>> instances are created first, then data read from Kafka consumer is
>> divided into 4 partitions (or something similar), and then map(record:
>> String) is called within something like while loop? Or what is the
>> actual flow? Or source code I can start from (I trace through
>> StreamExecutionEnvironment/ addSource/ DataStream/ transform/
>> addOperator etc., but I then get lost in source code)?
>>
>> Basically my problem is I have an AsyncHttpClient instance opened
>> within open() function and close in close function according to the
>> RichMapFunction doc. However, an issue is that in some cases my
>> AsyncHttpClient instance is not executed which displays warning like
>>
>> AsyncHttpClient.close() hasn't been invoked, which may produce file
>> descriptor leaks
>>
>> Therefore I would like to know the life cycle so that I can close
>> resource appropriately.
>>
>> Thanks
>
>