You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@beam.apache.org by shrikant bang <ma...@gmail.com> on 2021/01/10 18:13:18 UTC
Understanding KafkaIO.Read In Batch Mode
Hi Team,
I have below questions/ understandings for KafkaIO.Read in batch mode :
1. I built an understanding on debugging that, KafkaIO converts
unbounded stream into bounded read and *buffers all records* till either
of criteria matches - max records/ max time to read.
If this understanding is correct, then read is memory intensive as
KafkaIO has to buffer all read records before passing to down-streams. Is
my understanding correct?
2. If #1 is correct, then is there any way we can keep writing records
instead of buffering into memory in KafkaIO read operation (in batch mode) ?
Thank You,
Shrikant Bang
Re: Understanding KafkaIO.Read In Batch Mode
Posted by shrikant bang <ma...@gmail.com>.
Hi Boyaun,
Here is pipeline snippet of executed pipeline:
Map<String, Object> map = Maps.newHashMap();
map.put("enable.auto.commit", "false");
map.put("auto.offset.reset", "earliest");
map.put("group.id", options.getGroupId());
PTransform<PBegin, PCollection<KV<String, String>>> kafka =
KafkaIO.<String, String>read()
.withBootstrapServers(options.getBootstrap())
.withTopic(options.getTopic())
.withKeyDeserializer(StringDeserializer.class)
.withValueDeserializer(StringDeserializer.class)
.withMaxNumRecords(options.getKafkaRecordsToBeRead())
.commitOffsetsInFinalize()
.withConsumerConfigUpdates(map)
.commitOffsetsInFinalize()
.withoutMetadata();
pipeline.apply(kafka)
.apply(Values.<String>create())
.apply(TextIO.write().to(options.getOutput()));
pipeline.run().waitUntilFinish();
Thank You,
Shrikant Bang.
On Tue, Jan 12, 2021 at 3:40 PM shrikant bang <ma...@gmail.com>
wrote:
> Thanks, Boyuan!
>
> Here are details of the of KafkaIO.Read pipe:
>
> kafka-topic-partitions : 24
> max-records-to-be read : 15000 ( so total splits will be 2 -- min(100,
> (15000/10000 + 1))
> size of records between 300k-500k
> spark executor : 1 exec with 8g and 1 core
>
> With the above case, ETL pipeline has worked well. However, it seems
> KafkaIO read records from split will be added to DoFnOutputManager[1] into
> Multimap.
> I have attached a screen shot after adding a debug point at the iterator()
> method [2] ( size of multimap become 7500)
>
> If I increase max-records to be consumed from 15000 to 25000 then the
> number of splits will change from 2 to 3 and split fetch size changes from
> 7500 to ~8333 - This is making OOM for executors.
>
> I wonder if this is caused due to [1].
>
> [1]
> https://github.com/apache/beam/blob/master/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/MultiDoFnFunction.java#L244
> [2]
> https://github.com/apache/beam/blob/master/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/MultiDoFnFunction.java#L253
>
> [image: image.png]
>
> *P.S.* : I am testing with a single executor for debug purposes with 15k
> records.
>
> Thank You,
> Shrikant Bang
>
> On Tue, Jan 12, 2021 at 9:46 AM Boyuan Zhang <bo...@google.com> wrote:
>
>> Would you like to share the whole pipeline code here? With what kind of
>> configuration you will notice OOM, like how many elements, how many shards
>> and how many parallelisms you have for your run?
>>
>> On Mon, Jan 11, 2021 at 8:11 PM shrikant bang <
>> mailtoshrikant.bang@gmail.com> wrote:
>>
>>> Hi Boyuan,
>>>
>>> Thank you for your response. If I understood correctly, does
>>> each shard's output get batched into memory in DoFnOutputManager [1]
>>> before passing to downstream operations?
>>>
>>> I am trying to understand the root cause of OOM in executors if I
>>> increase the max number of records to be read without changing the
>>> executor's memory, numbers or cpu cores assigned.
>>>
>>> Below is a snippet for reading from Kafka:
>>>
>>> PTransform<PBegin, PCollection<KV<String, String>>> kafka =
>>> KafkaIO.<String, String>read()
>>> .withBootstrapServers(options.getBootstrap())
>>> .withTopic(options.getTopic())
>>> .withKeyDeserializer(StringDeserializer.class)
>>> .withValueDeserializer(StringDeserializer.class)
>>> .withMaxNumRecords(options.getKafkaRecordsToBeRead())
>>> .commitOffsetsInFinalize()
>>> .withConsumerConfigUpdates(map)
>>> .commitOffsetsInFinalize()
>>> .withoutMetadata();
>>>
>>>
>>> [1]
>>> https://github.com/apache/beam/blob/master/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/MultiDoFnFunction.java#L244
>>>
>>>
>>> Thank You,
>>> Shrikant Bang
>>>
>>>
>>> On Mon, Jan 11, 2021 at 11:54 PM Boyuan Zhang <bo...@google.com>
>>> wrote:
>>>
>>>> +dev <de...@beam.apache.org>
>>>>
>>>> Hi Shrikant,
>>>> If you look into the expansion of BoundedReadfromUnboundedSource[1],
>>>> you will notice that it will expand into Create single shard -> Split into
>>>> multiple shard -> read from one shard. The number of records from one shard
>>>> will not be larger than 10000 and the number of shards will not be larger
>>>> than 100[2].
>>>>
>>>> Back to your questions, the OutputReceiver will output the received
>>>> element to the downstream operation immediately. It will not keep the
>>>> record in batch in memory. But you are right that
>>>> BoundedReadFromUnboundedSource is memory-sensitive especially if your
>>>> records or your downstream operations consume a lot.
>>>>
>>>> [1]
>>>> https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L93-L121
>>>> [2]
>>>> https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L156-L161
>>>>
>>>>
>>>> On Mon, Jan 11, 2021 at 5:48 AM shrikant bang <
>>>> mailtoshrikant.bang@gmail.com> wrote:
>>>>
>>>>> Hi Team,
>>>>>
>>>>> I have ETL use cases with source as Kafka ( in *Batch* mode)
>>>>> with SparkRunner. I am trying to understand the internals' of
>>>>> KafkaIO.Read.
>>>>>
>>>>> Can someone please confirm if my understanding is correct?
>>>>>
>>>>> - WindowedContextOutputReceiver is getting used for collecting
>>>>> Kafka records from KafkaIO.Read from BoundedReadFromUnboundedSource [1].
>>>>> - All read Kafka records get stored in memory and gets spilled to
>>>>> downstream once the loop ends in Read function [2].
>>>>>
>>>>> Ref :
>>>>> [1] :
>>>>> https://github.com/apache/beam/blob/3bb232fb098700de408f574585dfe74bbaff7230/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L206
>>>>>
>>>>> [2] :
>>>>> https://github.com/apache/beam/blob/3bb232fb098700de408f574585dfe74bbaff7230/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L201
>>>>>
>>>>>
>>>>>
>>>>> Thank You,
>>>>> Shrikant Bang.
>>>>>
>>>>> On Sun, Jan 10, 2021 at 11:43 PM shrikant bang <
>>>>> mailtoshrikant.bang@gmail.com> wrote:
>>>>>
>>>>>> Hi Team,
>>>>>>
>>>>>> I have below questions/ understandings for KafkaIO.Read in batch
>>>>>> mode :
>>>>>>
>>>>>> 1. I built an understanding on debugging that, KafkaIO converts
>>>>>> unbounded stream into bounded read and *buffers all records* till
>>>>>> either of criteria matches - max records/ max time to read.
>>>>>> If this understanding is correct, then read is memory intensive
>>>>>> as KafkaIO has to buffer all read records before passing to down-streams.
>>>>>> Is my understanding correct?
>>>>>>
>>>>>> 2. If #1 is correct, then is there any way we can keep writing
>>>>>> records instead of buffering into memory in KafkaIO read operation (in
>>>>>> batch mode) ?
>>>>>>
>>>>>>
>>>>>> Thank You,
>>>>>> Shrikant Bang
>>>>>>
>>>>>
Re: Understanding KafkaIO.Read In Batch Mode
Posted by shrikant bang <ma...@gmail.com>.
Hi Boyaun,
Here is pipeline snippet of executed pipeline:
Map<String, Object> map = Maps.newHashMap();
map.put("enable.auto.commit", "false");
map.put("auto.offset.reset", "earliest");
map.put("group.id", options.getGroupId());
PTransform<PBegin, PCollection<KV<String, String>>> kafka =
KafkaIO.<String, String>read()
.withBootstrapServers(options.getBootstrap())
.withTopic(options.getTopic())
.withKeyDeserializer(StringDeserializer.class)
.withValueDeserializer(StringDeserializer.class)
.withMaxNumRecords(options.getKafkaRecordsToBeRead())
.commitOffsetsInFinalize()
.withConsumerConfigUpdates(map)
.commitOffsetsInFinalize()
.withoutMetadata();
pipeline.apply(kafka)
.apply(Values.<String>create())
.apply(TextIO.write().to(options.getOutput()));
pipeline.run().waitUntilFinish();
Thank You,
Shrikant Bang.
On Tue, Jan 12, 2021 at 3:40 PM shrikant bang <ma...@gmail.com>
wrote:
> Thanks, Boyuan!
>
> Here are details of the of KafkaIO.Read pipe:
>
> kafka-topic-partitions : 24
> max-records-to-be read : 15000 ( so total splits will be 2 -- min(100,
> (15000/10000 + 1))
> size of records between 300k-500k
> spark executor : 1 exec with 8g and 1 core
>
> With the above case, ETL pipeline has worked well. However, it seems
> KafkaIO read records from split will be added to DoFnOutputManager[1] into
> Multimap.
> I have attached a screen shot after adding a debug point at the iterator()
> method [2] ( size of multimap become 7500)
>
> If I increase max-records to be consumed from 15000 to 25000 then the
> number of splits will change from 2 to 3 and split fetch size changes from
> 7500 to ~8333 - This is making OOM for executors.
>
> I wonder if this is caused due to [1].
>
> [1]
> https://github.com/apache/beam/blob/master/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/MultiDoFnFunction.java#L244
> [2]
> https://github.com/apache/beam/blob/master/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/MultiDoFnFunction.java#L253
>
> [image: image.png]
>
> *P.S.* : I am testing with a single executor for debug purposes with 15k
> records.
>
> Thank You,
> Shrikant Bang
>
> On Tue, Jan 12, 2021 at 9:46 AM Boyuan Zhang <bo...@google.com> wrote:
>
>> Would you like to share the whole pipeline code here? With what kind of
>> configuration you will notice OOM, like how many elements, how many shards
>> and how many parallelisms you have for your run?
>>
>> On Mon, Jan 11, 2021 at 8:11 PM shrikant bang <
>> mailtoshrikant.bang@gmail.com> wrote:
>>
>>> Hi Boyuan,
>>>
>>> Thank you for your response. If I understood correctly, does
>>> each shard's output get batched into memory in DoFnOutputManager [1]
>>> before passing to downstream operations?
>>>
>>> I am trying to understand the root cause of OOM in executors if I
>>> increase the max number of records to be read without changing the
>>> executor's memory, numbers or cpu cores assigned.
>>>
>>> Below is a snippet for reading from Kafka:
>>>
>>> PTransform<PBegin, PCollection<KV<String, String>>> kafka =
>>> KafkaIO.<String, String>read()
>>> .withBootstrapServers(options.getBootstrap())
>>> .withTopic(options.getTopic())
>>> .withKeyDeserializer(StringDeserializer.class)
>>> .withValueDeserializer(StringDeserializer.class)
>>> .withMaxNumRecords(options.getKafkaRecordsToBeRead())
>>> .commitOffsetsInFinalize()
>>> .withConsumerConfigUpdates(map)
>>> .commitOffsetsInFinalize()
>>> .withoutMetadata();
>>>
>>>
>>> [1]
>>> https://github.com/apache/beam/blob/master/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/MultiDoFnFunction.java#L244
>>>
>>>
>>> Thank You,
>>> Shrikant Bang
>>>
>>>
>>> On Mon, Jan 11, 2021 at 11:54 PM Boyuan Zhang <bo...@google.com>
>>> wrote:
>>>
>>>> +dev <de...@beam.apache.org>
>>>>
>>>> Hi Shrikant,
>>>> If you look into the expansion of BoundedReadfromUnboundedSource[1],
>>>> you will notice that it will expand into Create single shard -> Split into
>>>> multiple shard -> read from one shard. The number of records from one shard
>>>> will not be larger than 10000 and the number of shards will not be larger
>>>> than 100[2].
>>>>
>>>> Back to your questions, the OutputReceiver will output the received
>>>> element to the downstream operation immediately. It will not keep the
>>>> record in batch in memory. But you are right that
>>>> BoundedReadFromUnboundedSource is memory-sensitive especially if your
>>>> records or your downstream operations consume a lot.
>>>>
>>>> [1]
>>>> https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L93-L121
>>>> [2]
>>>> https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L156-L161
>>>>
>>>>
>>>> On Mon, Jan 11, 2021 at 5:48 AM shrikant bang <
>>>> mailtoshrikant.bang@gmail.com> wrote:
>>>>
>>>>> Hi Team,
>>>>>
>>>>> I have ETL use cases with source as Kafka ( in *Batch* mode)
>>>>> with SparkRunner. I am trying to understand the internals' of
>>>>> KafkaIO.Read.
>>>>>
>>>>> Can someone please confirm if my understanding is correct?
>>>>>
>>>>> - WindowedContextOutputReceiver is getting used for collecting
>>>>> Kafka records from KafkaIO.Read from BoundedReadFromUnboundedSource [1].
>>>>> - All read Kafka records get stored in memory and gets spilled to
>>>>> downstream once the loop ends in Read function [2].
>>>>>
>>>>> Ref :
>>>>> [1] :
>>>>> https://github.com/apache/beam/blob/3bb232fb098700de408f574585dfe74bbaff7230/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L206
>>>>>
>>>>> [2] :
>>>>> https://github.com/apache/beam/blob/3bb232fb098700de408f574585dfe74bbaff7230/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L201
>>>>>
>>>>>
>>>>>
>>>>> Thank You,
>>>>> Shrikant Bang.
>>>>>
>>>>> On Sun, Jan 10, 2021 at 11:43 PM shrikant bang <
>>>>> mailtoshrikant.bang@gmail.com> wrote:
>>>>>
>>>>>> Hi Team,
>>>>>>
>>>>>> I have below questions/ understandings for KafkaIO.Read in batch
>>>>>> mode :
>>>>>>
>>>>>> 1. I built an understanding on debugging that, KafkaIO converts
>>>>>> unbounded stream into bounded read and *buffers all records* till
>>>>>> either of criteria matches - max records/ max time to read.
>>>>>> If this understanding is correct, then read is memory intensive
>>>>>> as KafkaIO has to buffer all read records before passing to down-streams.
>>>>>> Is my understanding correct?
>>>>>>
>>>>>> 2. If #1 is correct, then is there any way we can keep writing
>>>>>> records instead of buffering into memory in KafkaIO read operation (in
>>>>>> batch mode) ?
>>>>>>
>>>>>>
>>>>>> Thank You,
>>>>>> Shrikant Bang
>>>>>>
>>>>>
Re: Understanding KafkaIO.Read In Batch Mode
Posted by shrikant bang <ma...@gmail.com>.
Thanks, Boyuan!
Here are details of the of KafkaIO.Read pipe:
kafka-topic-partitions : 24
max-records-to-be read : 15000 ( so total splits will be 2 -- min(100,
(15000/10000 + 1))
size of records between 300k-500k
spark executor : 1 exec with 8g and 1 core
With the above case, ETL pipeline has worked well. However, it seems
KafkaIO read records from split will be added to DoFnOutputManager[1] into
Multimap.
I have attached a screen shot after adding a debug point at the iterator()
method [2] ( size of multimap become 7500)
If I increase max-records to be consumed from 15000 to 25000 then the
number of splits will change from 2 to 3 and split fetch size changes from
7500 to ~8333 - This is making OOM for executors.
I wonder if this is caused due to [1].
[1]
https://github.com/apache/beam/blob/master/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/MultiDoFnFunction.java#L244
[2]
https://github.com/apache/beam/blob/master/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/MultiDoFnFunction.java#L253
[image: image.png]
*P.S.* : I am testing with a single executor for debug purposes with 15k
records.
Thank You,
Shrikant Bang
On Tue, Jan 12, 2021 at 9:46 AM Boyuan Zhang <bo...@google.com> wrote:
> Would you like to share the whole pipeline code here? With what kind of
> configuration you will notice OOM, like how many elements, how many shards
> and how many parallelisms you have for your run?
>
> On Mon, Jan 11, 2021 at 8:11 PM shrikant bang <
> mailtoshrikant.bang@gmail.com> wrote:
>
>> Hi Boyuan,
>>
>> Thank you for your response. If I understood correctly, does each
>> shard's output get batched into memory in DoFnOutputManager [1] before
>> passing to downstream operations?
>>
>> I am trying to understand the root cause of OOM in executors if I
>> increase the max number of records to be read without changing the
>> executor's memory, numbers or cpu cores assigned.
>>
>> Below is a snippet for reading from Kafka:
>>
>> PTransform<PBegin, PCollection<KV<String, String>>> kafka =
>> KafkaIO.<String, String>read()
>> .withBootstrapServers(options.getBootstrap())
>> .withTopic(options.getTopic())
>> .withKeyDeserializer(StringDeserializer.class)
>> .withValueDeserializer(StringDeserializer.class)
>> .withMaxNumRecords(options.getKafkaRecordsToBeRead())
>> .commitOffsetsInFinalize()
>> .withConsumerConfigUpdates(map)
>> .commitOffsetsInFinalize()
>> .withoutMetadata();
>>
>>
>> [1]
>> https://github.com/apache/beam/blob/master/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/MultiDoFnFunction.java#L244
>>
>>
>> Thank You,
>> Shrikant Bang
>>
>>
>> On Mon, Jan 11, 2021 at 11:54 PM Boyuan Zhang <bo...@google.com> wrote:
>>
>>> +dev <de...@beam.apache.org>
>>>
>>> Hi Shrikant,
>>> If you look into the expansion of BoundedReadfromUnboundedSource[1], you
>>> will notice that it will expand into Create single shard -> Split into
>>> multiple shard -> read from one shard. The number of records from one shard
>>> will not be larger than 10000 and the number of shards will not be larger
>>> than 100[2].
>>>
>>> Back to your questions, the OutputReceiver will output the received
>>> element to the downstream operation immediately. It will not keep the
>>> record in batch in memory. But you are right that
>>> BoundedReadFromUnboundedSource is memory-sensitive especially if your
>>> records or your downstream operations consume a lot.
>>>
>>> [1]
>>> https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L93-L121
>>> [2]
>>> https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L156-L161
>>>
>>>
>>> On Mon, Jan 11, 2021 at 5:48 AM shrikant bang <
>>> mailtoshrikant.bang@gmail.com> wrote:
>>>
>>>> Hi Team,
>>>>
>>>> I have ETL use cases with source as Kafka ( in *Batch* mode)
>>>> with SparkRunner. I am trying to understand the internals' of
>>>> KafkaIO.Read.
>>>>
>>>> Can someone please confirm if my understanding is correct?
>>>>
>>>> - WindowedContextOutputReceiver is getting used for collecting Kafka
>>>> records from KafkaIO.Read from BoundedReadFromUnboundedSource [1].
>>>> - All read Kafka records get stored in memory and gets spilled to
>>>> downstream once the loop ends in Read function [2].
>>>>
>>>> Ref :
>>>> [1] :
>>>> https://github.com/apache/beam/blob/3bb232fb098700de408f574585dfe74bbaff7230/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L206
>>>>
>>>> [2] :
>>>> https://github.com/apache/beam/blob/3bb232fb098700de408f574585dfe74bbaff7230/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L201
>>>>
>>>>
>>>>
>>>> Thank You,
>>>> Shrikant Bang.
>>>>
>>>> On Sun, Jan 10, 2021 at 11:43 PM shrikant bang <
>>>> mailtoshrikant.bang@gmail.com> wrote:
>>>>
>>>>> Hi Team,
>>>>>
>>>>> I have below questions/ understandings for KafkaIO.Read in batch
>>>>> mode :
>>>>>
>>>>> 1. I built an understanding on debugging that, KafkaIO converts
>>>>> unbounded stream into bounded read and *buffers all records* till
>>>>> either of criteria matches - max records/ max time to read.
>>>>> If this understanding is correct, then read is memory intensive as
>>>>> KafkaIO has to buffer all read records before passing to down-streams. Is
>>>>> my understanding correct?
>>>>>
>>>>> 2. If #1 is correct, then is there any way we can keep writing
>>>>> records instead of buffering into memory in KafkaIO read operation (in
>>>>> batch mode) ?
>>>>>
>>>>>
>>>>> Thank You,
>>>>> Shrikant Bang
>>>>>
>>>>
Re: Understanding KafkaIO.Read In Batch Mode
Posted by shrikant bang <ma...@gmail.com>.
Thanks, Boyuan!
Here are details of the of KafkaIO.Read pipe:
kafka-topic-partitions : 24
max-records-to-be read : 15000 ( so total splits will be 2 -- min(100,
(15000/10000 + 1))
size of records between 300k-500k
spark executor : 1 exec with 8g and 1 core
With the above case, ETL pipeline has worked well. However, it seems
KafkaIO read records from split will be added to DoFnOutputManager[1] into
Multimap.
I have attached a screen shot after adding a debug point at the iterator()
method [2] ( size of multimap become 7500)
If I increase max-records to be consumed from 15000 to 25000 then the
number of splits will change from 2 to 3 and split fetch size changes from
7500 to ~8333 - This is making OOM for executors.
I wonder if this is caused due to [1].
[1]
https://github.com/apache/beam/blob/master/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/MultiDoFnFunction.java#L244
[2]
https://github.com/apache/beam/blob/master/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/MultiDoFnFunction.java#L253
[image: image.png]
*P.S.* : I am testing with a single executor for debug purposes with 15k
records.
Thank You,
Shrikant Bang
On Tue, Jan 12, 2021 at 9:46 AM Boyuan Zhang <bo...@google.com> wrote:
> Would you like to share the whole pipeline code here? With what kind of
> configuration you will notice OOM, like how many elements, how many shards
> and how many parallelisms you have for your run?
>
> On Mon, Jan 11, 2021 at 8:11 PM shrikant bang <
> mailtoshrikant.bang@gmail.com> wrote:
>
>> Hi Boyuan,
>>
>> Thank you for your response. If I understood correctly, does each
>> shard's output get batched into memory in DoFnOutputManager [1] before
>> passing to downstream operations?
>>
>> I am trying to understand the root cause of OOM in executors if I
>> increase the max number of records to be read without changing the
>> executor's memory, numbers or cpu cores assigned.
>>
>> Below is a snippet for reading from Kafka:
>>
>> PTransform<PBegin, PCollection<KV<String, String>>> kafka =
>> KafkaIO.<String, String>read()
>> .withBootstrapServers(options.getBootstrap())
>> .withTopic(options.getTopic())
>> .withKeyDeserializer(StringDeserializer.class)
>> .withValueDeserializer(StringDeserializer.class)
>> .withMaxNumRecords(options.getKafkaRecordsToBeRead())
>> .commitOffsetsInFinalize()
>> .withConsumerConfigUpdates(map)
>> .commitOffsetsInFinalize()
>> .withoutMetadata();
>>
>>
>> [1]
>> https://github.com/apache/beam/blob/master/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/MultiDoFnFunction.java#L244
>>
>>
>> Thank You,
>> Shrikant Bang
>>
>>
>> On Mon, Jan 11, 2021 at 11:54 PM Boyuan Zhang <bo...@google.com> wrote:
>>
>>> +dev <de...@beam.apache.org>
>>>
>>> Hi Shrikant,
>>> If you look into the expansion of BoundedReadfromUnboundedSource[1], you
>>> will notice that it will expand into Create single shard -> Split into
>>> multiple shard -> read from one shard. The number of records from one shard
>>> will not be larger than 10000 and the number of shards will not be larger
>>> than 100[2].
>>>
>>> Back to your questions, the OutputReceiver will output the received
>>> element to the downstream operation immediately. It will not keep the
>>> record in batch in memory. But you are right that
>>> BoundedReadFromUnboundedSource is memory-sensitive especially if your
>>> records or your downstream operations consume a lot.
>>>
>>> [1]
>>> https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L93-L121
>>> [2]
>>> https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L156-L161
>>>
>>>
>>> On Mon, Jan 11, 2021 at 5:48 AM shrikant bang <
>>> mailtoshrikant.bang@gmail.com> wrote:
>>>
>>>> Hi Team,
>>>>
>>>> I have ETL use cases with source as Kafka ( in *Batch* mode)
>>>> with SparkRunner. I am trying to understand the internals' of
>>>> KafkaIO.Read.
>>>>
>>>> Can someone please confirm if my understanding is correct?
>>>>
>>>> - WindowedContextOutputReceiver is getting used for collecting Kafka
>>>> records from KafkaIO.Read from BoundedReadFromUnboundedSource [1].
>>>> - All read Kafka records get stored in memory and gets spilled to
>>>> downstream once the loop ends in Read function [2].
>>>>
>>>> Ref :
>>>> [1] :
>>>> https://github.com/apache/beam/blob/3bb232fb098700de408f574585dfe74bbaff7230/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L206
>>>>
>>>> [2] :
>>>> https://github.com/apache/beam/blob/3bb232fb098700de408f574585dfe74bbaff7230/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L201
>>>>
>>>>
>>>>
>>>> Thank You,
>>>> Shrikant Bang.
>>>>
>>>> On Sun, Jan 10, 2021 at 11:43 PM shrikant bang <
>>>> mailtoshrikant.bang@gmail.com> wrote:
>>>>
>>>>> Hi Team,
>>>>>
>>>>> I have below questions/ understandings for KafkaIO.Read in batch
>>>>> mode :
>>>>>
>>>>> 1. I built an understanding on debugging that, KafkaIO converts
>>>>> unbounded stream into bounded read and *buffers all records* till
>>>>> either of criteria matches - max records/ max time to read.
>>>>> If this understanding is correct, then read is memory intensive as
>>>>> KafkaIO has to buffer all read records before passing to down-streams. Is
>>>>> my understanding correct?
>>>>>
>>>>> 2. If #1 is correct, then is there any way we can keep writing
>>>>> records instead of buffering into memory in KafkaIO read operation (in
>>>>> batch mode) ?
>>>>>
>>>>>
>>>>> Thank You,
>>>>> Shrikant Bang
>>>>>
>>>>
Re: Understanding KafkaIO.Read In Batch Mode
Posted by Boyuan Zhang <bo...@google.com>.
Would you like to share the whole pipeline code here? With what kind of
configuration you will notice OOM, like how many elements, how many shards
and how many parallelisms you have for your run?
On Mon, Jan 11, 2021 at 8:11 PM shrikant bang <ma...@gmail.com>
wrote:
> Hi Boyuan,
>
> Thank you for your response. If I understood correctly, does each
> shard's output get batched into memory in DoFnOutputManager [1] before
> passing to downstream operations?
>
> I am trying to understand the root cause of OOM in executors if I
> increase the max number of records to be read without changing the
> executor's memory, numbers or cpu cores assigned.
>
> Below is a snippet for reading from Kafka:
>
> PTransform<PBegin, PCollection<KV<String, String>>> kafka =
> KafkaIO.<String, String>read()
> .withBootstrapServers(options.getBootstrap())
> .withTopic(options.getTopic())
> .withKeyDeserializer(StringDeserializer.class)
> .withValueDeserializer(StringDeserializer.class)
> .withMaxNumRecords(options.getKafkaRecordsToBeRead())
> .commitOffsetsInFinalize()
> .withConsumerConfigUpdates(map)
> .commitOffsetsInFinalize()
> .withoutMetadata();
>
>
> [1]
> https://github.com/apache/beam/blob/master/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/MultiDoFnFunction.java#L244
>
>
> Thank You,
> Shrikant Bang
>
>
> On Mon, Jan 11, 2021 at 11:54 PM Boyuan Zhang <bo...@google.com> wrote:
>
>> +dev <de...@beam.apache.org>
>>
>> Hi Shrikant,
>> If you look into the expansion of BoundedReadfromUnboundedSource[1], you
>> will notice that it will expand into Create single shard -> Split into
>> multiple shard -> read from one shard. The number of records from one shard
>> will not be larger than 10000 and the number of shards will not be larger
>> than 100[2].
>>
>> Back to your questions, the OutputReceiver will output the received
>> element to the downstream operation immediately. It will not keep the
>> record in batch in memory. But you are right that
>> BoundedReadFromUnboundedSource is memory-sensitive especially if your
>> records or your downstream operations consume a lot.
>>
>> [1]
>> https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L93-L121
>> [2]
>> https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L156-L161
>>
>>
>> On Mon, Jan 11, 2021 at 5:48 AM shrikant bang <
>> mailtoshrikant.bang@gmail.com> wrote:
>>
>>> Hi Team,
>>>
>>> I have ETL use cases with source as Kafka ( in *Batch* mode)
>>> with SparkRunner. I am trying to understand the internals' of
>>> KafkaIO.Read.
>>>
>>> Can someone please confirm if my understanding is correct?
>>>
>>> - WindowedContextOutputReceiver is getting used for collecting Kafka
>>> records from KafkaIO.Read from BoundedReadFromUnboundedSource [1].
>>> - All read Kafka records get stored in memory and gets spilled to
>>> downstream once the loop ends in Read function [2].
>>>
>>> Ref :
>>> [1] :
>>> https://github.com/apache/beam/blob/3bb232fb098700de408f574585dfe74bbaff7230/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L206
>>>
>>> [2] :
>>> https://github.com/apache/beam/blob/3bb232fb098700de408f574585dfe74bbaff7230/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L201
>>>
>>>
>>>
>>> Thank You,
>>> Shrikant Bang.
>>>
>>> On Sun, Jan 10, 2021 at 11:43 PM shrikant bang <
>>> mailtoshrikant.bang@gmail.com> wrote:
>>>
>>>> Hi Team,
>>>>
>>>> I have below questions/ understandings for KafkaIO.Read in batch
>>>> mode :
>>>>
>>>> 1. I built an understanding on debugging that, KafkaIO converts
>>>> unbounded stream into bounded read and *buffers all records* till
>>>> either of criteria matches - max records/ max time to read.
>>>> If this understanding is correct, then read is memory intensive as
>>>> KafkaIO has to buffer all read records before passing to down-streams. Is
>>>> my understanding correct?
>>>>
>>>> 2. If #1 is correct, then is there any way we can keep writing
>>>> records instead of buffering into memory in KafkaIO read operation (in
>>>> batch mode) ?
>>>>
>>>>
>>>> Thank You,
>>>> Shrikant Bang
>>>>
>>>
Re: Understanding KafkaIO.Read In Batch Mode
Posted by Boyuan Zhang <bo...@google.com>.
Would you like to share the whole pipeline code here? With what kind of
configuration you will notice OOM, like how many elements, how many shards
and how many parallelisms you have for your run?
On Mon, Jan 11, 2021 at 8:11 PM shrikant bang <ma...@gmail.com>
wrote:
> Hi Boyuan,
>
> Thank you for your response. If I understood correctly, does each
> shard's output get batched into memory in DoFnOutputManager [1] before
> passing to downstream operations?
>
> I am trying to understand the root cause of OOM in executors if I
> increase the max number of records to be read without changing the
> executor's memory, numbers or cpu cores assigned.
>
> Below is a snippet for reading from Kafka:
>
> PTransform<PBegin, PCollection<KV<String, String>>> kafka =
> KafkaIO.<String, String>read()
> .withBootstrapServers(options.getBootstrap())
> .withTopic(options.getTopic())
> .withKeyDeserializer(StringDeserializer.class)
> .withValueDeserializer(StringDeserializer.class)
> .withMaxNumRecords(options.getKafkaRecordsToBeRead())
> .commitOffsetsInFinalize()
> .withConsumerConfigUpdates(map)
> .commitOffsetsInFinalize()
> .withoutMetadata();
>
>
> [1]
> https://github.com/apache/beam/blob/master/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/MultiDoFnFunction.java#L244
>
>
> Thank You,
> Shrikant Bang
>
>
> On Mon, Jan 11, 2021 at 11:54 PM Boyuan Zhang <bo...@google.com> wrote:
>
>> +dev <de...@beam.apache.org>
>>
>> Hi Shrikant,
>> If you look into the expansion of BoundedReadfromUnboundedSource[1], you
>> will notice that it will expand into Create single shard -> Split into
>> multiple shard -> read from one shard. The number of records from one shard
>> will not be larger than 10000 and the number of shards will not be larger
>> than 100[2].
>>
>> Back to your questions, the OutputReceiver will output the received
>> element to the downstream operation immediately. It will not keep the
>> record in batch in memory. But you are right that
>> BoundedReadFromUnboundedSource is memory-sensitive especially if your
>> records or your downstream operations consume a lot.
>>
>> [1]
>> https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L93-L121
>> [2]
>> https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L156-L161
>>
>>
>> On Mon, Jan 11, 2021 at 5:48 AM shrikant bang <
>> mailtoshrikant.bang@gmail.com> wrote:
>>
>>> Hi Team,
>>>
>>> I have ETL use cases with source as Kafka ( in *Batch* mode)
>>> with SparkRunner. I am trying to understand the internals' of
>>> KafkaIO.Read.
>>>
>>> Can someone please confirm if my understanding is correct?
>>>
>>> - WindowedContextOutputReceiver is getting used for collecting Kafka
>>> records from KafkaIO.Read from BoundedReadFromUnboundedSource [1].
>>> - All read Kafka records get stored in memory and gets spilled to
>>> downstream once the loop ends in Read function [2].
>>>
>>> Ref :
>>> [1] :
>>> https://github.com/apache/beam/blob/3bb232fb098700de408f574585dfe74bbaff7230/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L206
>>>
>>> [2] :
>>> https://github.com/apache/beam/blob/3bb232fb098700de408f574585dfe74bbaff7230/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L201
>>>
>>>
>>>
>>> Thank You,
>>> Shrikant Bang.
>>>
>>> On Sun, Jan 10, 2021 at 11:43 PM shrikant bang <
>>> mailtoshrikant.bang@gmail.com> wrote:
>>>
>>>> Hi Team,
>>>>
>>>> I have below questions/ understandings for KafkaIO.Read in batch
>>>> mode :
>>>>
>>>> 1. I built an understanding on debugging that, KafkaIO converts
>>>> unbounded stream into bounded read and *buffers all records* till
>>>> either of criteria matches - max records/ max time to read.
>>>> If this understanding is correct, then read is memory intensive as
>>>> KafkaIO has to buffer all read records before passing to down-streams. Is
>>>> my understanding correct?
>>>>
>>>> 2. If #1 is correct, then is there any way we can keep writing
>>>> records instead of buffering into memory in KafkaIO read operation (in
>>>> batch mode) ?
>>>>
>>>>
>>>> Thank You,
>>>> Shrikant Bang
>>>>
>>>
Re: Understanding KafkaIO.Read In Batch Mode
Posted by shrikant bang <ma...@gmail.com>.
Hi Boyuan,
Thank you for your response. If I understood correctly, does each
shard's output get batched into memory in DoFnOutputManager [1] before
passing to downstream operations?
I am trying to understand the root cause of OOM in executors if I
increase the max number of records to be read without changing the
executor's memory, numbers or cpu cores assigned.
Below is a snippet for reading from Kafka:
PTransform<PBegin, PCollection<KV<String, String>>> kafka =
KafkaIO.<String, String>read()
.withBootstrapServers(options.getBootstrap())
.withTopic(options.getTopic())
.withKeyDeserializer(StringDeserializer.class)
.withValueDeserializer(StringDeserializer.class)
.withMaxNumRecords(options.getKafkaRecordsToBeRead())
.commitOffsetsInFinalize()
.withConsumerConfigUpdates(map)
.commitOffsetsInFinalize()
.withoutMetadata();
[1]
https://github.com/apache/beam/blob/master/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/MultiDoFnFunction.java#L244
Thank You,
Shrikant Bang
On Mon, Jan 11, 2021 at 11:54 PM Boyuan Zhang <bo...@google.com> wrote:
> +dev <de...@beam.apache.org>
>
> Hi Shrikant,
> If you look into the expansion of BoundedReadfromUnboundedSource[1], you
> will notice that it will expand into Create single shard -> Split into
> multiple shard -> read from one shard. The number of records from one shard
> will not be larger than 10000 and the number of shards will not be larger
> than 100[2].
>
> Back to your questions, the OutputReceiver will output the received
> element to the downstream operation immediately. It will not keep the
> record in batch in memory. But you are right that
> BoundedReadFromUnboundedSource is memory-sensitive especially if your
> records or your downstream operations consume a lot.
>
> [1]
> https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L93-L121
> [2]
> https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L156-L161
>
>
> On Mon, Jan 11, 2021 at 5:48 AM shrikant bang <
> mailtoshrikant.bang@gmail.com> wrote:
>
>> Hi Team,
>>
>> I have ETL use cases with source as Kafka ( in *Batch* mode)
>> with SparkRunner. I am trying to understand the internals' of
>> KafkaIO.Read.
>>
>> Can someone please confirm if my understanding is correct?
>>
>> - WindowedContextOutputReceiver is getting used for collecting Kafka
>> records from KafkaIO.Read from BoundedReadFromUnboundedSource [1].
>> - All read Kafka records get stored in memory and gets spilled to
>> downstream once the loop ends in Read function [2].
>>
>> Ref :
>> [1] :
>> https://github.com/apache/beam/blob/3bb232fb098700de408f574585dfe74bbaff7230/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L206
>>
>> [2] :
>> https://github.com/apache/beam/blob/3bb232fb098700de408f574585dfe74bbaff7230/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L201
>>
>>
>>
>> Thank You,
>> Shrikant Bang.
>>
>> On Sun, Jan 10, 2021 at 11:43 PM shrikant bang <
>> mailtoshrikant.bang@gmail.com> wrote:
>>
>>> Hi Team,
>>>
>>> I have below questions/ understandings for KafkaIO.Read in batch mode
>>> :
>>>
>>> 1. I built an understanding on debugging that, KafkaIO converts
>>> unbounded stream into bounded read and *buffers all records* till
>>> either of criteria matches - max records/ max time to read.
>>> If this understanding is correct, then read is memory intensive as
>>> KafkaIO has to buffer all read records before passing to down-streams. Is
>>> my understanding correct?
>>>
>>> 2. If #1 is correct, then is there any way we can keep writing
>>> records instead of buffering into memory in KafkaIO read operation (in
>>> batch mode) ?
>>>
>>>
>>> Thank You,
>>> Shrikant Bang
>>>
>>
Re: Understanding KafkaIO.Read In Batch Mode
Posted by shrikant bang <ma...@gmail.com>.
Hi Boyuan,
Thank you for your response. If I understood correctly, does each
shard's output get batched into memory in DoFnOutputManager [1] before
passing to downstream operations?
I am trying to understand the root cause of OOM in executors if I
increase the max number of records to be read without changing the
executor's memory, numbers or cpu cores assigned.
Below is a snippet for reading from Kafka:
PTransform<PBegin, PCollection<KV<String, String>>> kafka =
KafkaIO.<String, String>read()
.withBootstrapServers(options.getBootstrap())
.withTopic(options.getTopic())
.withKeyDeserializer(StringDeserializer.class)
.withValueDeserializer(StringDeserializer.class)
.withMaxNumRecords(options.getKafkaRecordsToBeRead())
.commitOffsetsInFinalize()
.withConsumerConfigUpdates(map)
.commitOffsetsInFinalize()
.withoutMetadata();
[1]
https://github.com/apache/beam/blob/master/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/MultiDoFnFunction.java#L244
Thank You,
Shrikant Bang
On Mon, Jan 11, 2021 at 11:54 PM Boyuan Zhang <bo...@google.com> wrote:
> +dev <de...@beam.apache.org>
>
> Hi Shrikant,
> If you look into the expansion of BoundedReadfromUnboundedSource[1], you
> will notice that it will expand into Create single shard -> Split into
> multiple shard -> read from one shard. The number of records from one shard
> will not be larger than 10000 and the number of shards will not be larger
> than 100[2].
>
> Back to your questions, the OutputReceiver will output the received
> element to the downstream operation immediately. It will not keep the
> record in batch in memory. But you are right that
> BoundedReadFromUnboundedSource is memory-sensitive especially if your
> records or your downstream operations consume a lot.
>
> [1]
> https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L93-L121
> [2]
> https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L156-L161
>
>
> On Mon, Jan 11, 2021 at 5:48 AM shrikant bang <
> mailtoshrikant.bang@gmail.com> wrote:
>
>> Hi Team,
>>
>> I have ETL use cases with source as Kafka ( in *Batch* mode)
>> with SparkRunner. I am trying to understand the internals' of
>> KafkaIO.Read.
>>
>> Can someone please confirm if my understanding is correct?
>>
>> - WindowedContextOutputReceiver is getting used for collecting Kafka
>> records from KafkaIO.Read from BoundedReadFromUnboundedSource [1].
>> - All read Kafka records get stored in memory and gets spilled to
>> downstream once the loop ends in Read function [2].
>>
>> Ref :
>> [1] :
>> https://github.com/apache/beam/blob/3bb232fb098700de408f574585dfe74bbaff7230/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L206
>>
>> [2] :
>> https://github.com/apache/beam/blob/3bb232fb098700de408f574585dfe74bbaff7230/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L201
>>
>>
>>
>> Thank You,
>> Shrikant Bang.
>>
>> On Sun, Jan 10, 2021 at 11:43 PM shrikant bang <
>> mailtoshrikant.bang@gmail.com> wrote:
>>
>>> Hi Team,
>>>
>>> I have below questions/ understandings for KafkaIO.Read in batch mode
>>> :
>>>
>>> 1. I built an understanding on debugging that, KafkaIO converts
>>> unbounded stream into bounded read and *buffers all records* till
>>> either of criteria matches - max records/ max time to read.
>>> If this understanding is correct, then read is memory intensive as
>>> KafkaIO has to buffer all read records before passing to down-streams. Is
>>> my understanding correct?
>>>
>>> 2. If #1 is correct, then is there any way we can keep writing
>>> records instead of buffering into memory in KafkaIO read operation (in
>>> batch mode) ?
>>>
>>>
>>> Thank You,
>>> Shrikant Bang
>>>
>>
Re: Understanding KafkaIO.Read In Batch Mode
Posted by Boyuan Zhang <bo...@google.com>.
+dev <de...@beam.apache.org>
Hi Shrikant,
If you look into the expansion of BoundedReadfromUnboundedSource[1], you
will notice that it will expand into Create single shard -> Split into
multiple shard -> read from one shard. The number of records from one shard
will not be larger than 10000 and the number of shards will not be larger
than 100[2].
Back to your questions, the OutputReceiver will output the received
element to the downstream operation immediately. It will not keep the
record in batch in memory. But you are right that
BoundedReadFromUnboundedSource is memory-sensitive especially if your
records or your downstream operations consume a lot.
[1]
https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L93-L121
[2]
https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L156-L161
On Mon, Jan 11, 2021 at 5:48 AM shrikant bang <ma...@gmail.com>
wrote:
> Hi Team,
>
> I have ETL use cases with source as Kafka ( in *Batch* mode)
> with SparkRunner. I am trying to understand the internals' of
> KafkaIO.Read.
>
> Can someone please confirm if my understanding is correct?
>
> - WindowedContextOutputReceiver is getting used for collecting Kafka
> records from KafkaIO.Read from BoundedReadFromUnboundedSource [1].
> - All read Kafka records get stored in memory and gets spilled to
> downstream once the loop ends in Read function [2].
>
> Ref :
> [1] :
> https://github.com/apache/beam/blob/3bb232fb098700de408f574585dfe74bbaff7230/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L206
>
> [2] :
> https://github.com/apache/beam/blob/3bb232fb098700de408f574585dfe74bbaff7230/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L201
>
>
>
> Thank You,
> Shrikant Bang.
>
> On Sun, Jan 10, 2021 at 11:43 PM shrikant bang <
> mailtoshrikant.bang@gmail.com> wrote:
>
>> Hi Team,
>>
>> I have below questions/ understandings for KafkaIO.Read in batch mode :
>>
>> 1. I built an understanding on debugging that, KafkaIO converts
>> unbounded stream into bounded read and *buffers all records* till
>> either of criteria matches - max records/ max time to read.
>> If this understanding is correct, then read is memory intensive as
>> KafkaIO has to buffer all read records before passing to down-streams. Is
>> my understanding correct?
>>
>> 2. If #1 is correct, then is there any way we can keep writing
>> records instead of buffering into memory in KafkaIO read operation (in
>> batch mode) ?
>>
>>
>> Thank You,
>> Shrikant Bang
>>
>
Re: Understanding KafkaIO.Read In Batch Mode
Posted by Boyuan Zhang <bo...@google.com>.
+dev <de...@beam.apache.org>
Hi Shrikant,
If you look into the expansion of BoundedReadfromUnboundedSource[1], you
will notice that it will expand into Create single shard -> Split into
multiple shard -> read from one shard. The number of records from one shard
will not be larger than 10000 and the number of shards will not be larger
than 100[2].
Back to your questions, the OutputReceiver will output the received
element to the downstream operation immediately. It will not keep the
record in batch in memory. But you are right that
BoundedReadFromUnboundedSource is memory-sensitive especially if your
records or your downstream operations consume a lot.
[1]
https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L93-L121
[2]
https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L156-L161
On Mon, Jan 11, 2021 at 5:48 AM shrikant bang <ma...@gmail.com>
wrote:
> Hi Team,
>
> I have ETL use cases with source as Kafka ( in *Batch* mode)
> with SparkRunner. I am trying to understand the internals' of
> KafkaIO.Read.
>
> Can someone please confirm if my understanding is correct?
>
> - WindowedContextOutputReceiver is getting used for collecting Kafka
> records from KafkaIO.Read from BoundedReadFromUnboundedSource [1].
> - All read Kafka records get stored in memory and gets spilled to
> downstream once the loop ends in Read function [2].
>
> Ref :
> [1] :
> https://github.com/apache/beam/blob/3bb232fb098700de408f574585dfe74bbaff7230/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L206
>
> [2] :
> https://github.com/apache/beam/blob/3bb232fb098700de408f574585dfe74bbaff7230/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L201
>
>
>
> Thank You,
> Shrikant Bang.
>
> On Sun, Jan 10, 2021 at 11:43 PM shrikant bang <
> mailtoshrikant.bang@gmail.com> wrote:
>
>> Hi Team,
>>
>> I have below questions/ understandings for KafkaIO.Read in batch mode :
>>
>> 1. I built an understanding on debugging that, KafkaIO converts
>> unbounded stream into bounded read and *buffers all records* till
>> either of criteria matches - max records/ max time to read.
>> If this understanding is correct, then read is memory intensive as
>> KafkaIO has to buffer all read records before passing to down-streams. Is
>> my understanding correct?
>>
>> 2. If #1 is correct, then is there any way we can keep writing
>> records instead of buffering into memory in KafkaIO read operation (in
>> batch mode) ?
>>
>>
>> Thank You,
>> Shrikant Bang
>>
>
Re: Understanding KafkaIO.Read In Batch Mode
Posted by shrikant bang <ma...@gmail.com>.
Hi Team,
I have ETL use cases with source as Kafka ( in *Batch* mode)
with SparkRunner. I am trying to understand the internals' of
KafkaIO.Read.
Can someone please confirm if my understanding is correct?
- WindowedContextOutputReceiver is getting used for collecting Kafka
records from KafkaIO.Read from BoundedReadFromUnboundedSource [1].
- All read Kafka records get stored in memory and gets spilled to
downstream once the loop ends in Read function [2].
Ref :
[1] :
https://github.com/apache/beam/blob/3bb232fb098700de408f574585dfe74bbaff7230/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L206
[2] :
https://github.com/apache/beam/blob/3bb232fb098700de408f574585dfe74bbaff7230/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java#L201
Thank You,
Shrikant Bang.
On Sun, Jan 10, 2021 at 11:43 PM shrikant bang <
mailtoshrikant.bang@gmail.com> wrote:
> Hi Team,
>
> I have below questions/ understandings for KafkaIO.Read in batch mode :
>
> 1. I built an understanding on debugging that, KafkaIO converts
> unbounded stream into bounded read and *buffers all records* till
> either of criteria matches - max records/ max time to read.
> If this understanding is correct, then read is memory intensive as
> KafkaIO has to buffer all read records before passing to down-streams. Is
> my understanding correct?
>
> 2. If #1 is correct, then is there any way we can keep writing records
> instead of buffering into memory in KafkaIO read operation (in batch mode) ?
>
>
> Thank You,
> Shrikant Bang
>