You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@beam.apache.org by Jaehyeon Kim <do...@gmail.com> on 2024/07/02 10:27:42 UTC
KafkaIO no longer allows use_deprecated_read?
Hello,
A while ago, I deployed a beam pipeline on a Flink cluster (Flink 1.16.3)
and the DAG only had two tasks.
[image: image.png]
When I deployed a similar pipeline on Flink 1.18.1, the DAG looked much
more complicated.
[image: image.png]
In both the cases, I specified --experiment=use_deprecated_read.
Is there a change in the KafkaIO configuration or I may have made a mistake
to specifiy the pipeline argument?
Cheers,
Jaehyeon
Re: KafkaIO no longer allows use_deprecated_read?
Posted by Ahmet Altay via user <us...@beam.apache.org>.
Thank you. Filed an issue for this here:
https://github.com/apache/beam/issues/31839
On Wed, Jul 10, 2024 at 3:07 PM Jaehyeon Kim <do...@gmail.com> wrote:
> I don't think it is documented. Initially I noticed it in a post (
> https://ndeepak.com/posts/2022-07-07-local-beam/), and it refers to a
> GitHub issue (https://github.com/apache/beam/issues/20979). I followed
> both the examples but didn't work until I added it directly to the java io
> expansion service.
>
> On Thu, 11 Jul 2024 at 01:46, Ahmet Altay via user <us...@beam.apache.org>
> wrote:
>
>> Thank you Jaehyeon.
>>
>> Was this (setting experiments via expansion_service) documented?
>>
>> On Wed, Jul 10, 2024 at 1:53 AM Bartosz Zabłocki via user <
>> user@beam.apache.org> wrote:
>>
>>> Hi Jaehyeon,
>>> thanks for the update and the instructions on how to do this! I wasn't
>>> aware of the `expansion_service` argument.
>>>
>>> Cheers,
>>> Bartosz
>>>
>>> On Wed, Jul 10, 2024 at 6:04 AM Jaehyeon Kim <do...@gmail.com> wrote:
>>>
>>>> Hello,
>>>>
>>>> Looks like I have to specify that option in the Kafka IO methods rather
>>>> than specifying it as a pipeline argument. eg)
>>>>
>>>> >> kafka.ReadFromKafka(
>>>> consumer_config={
>>>> "bootstrap.servers": self.boostrap_servers,
>>>> "auto.offset.reset": "earliest",
>>>> # "enable.auto.commit": "true",
>>>> "group.id": self.group_id,
>>>> },
>>>> topics=self.topics,
>>>> timestamp_policy=kafka.ReadFromKafka.create_time_policy
>>>> ,
>>>> expansion_service=kafka.default_io_expansion_service(
>>>> ["--experiments=use_deprecated_read"]
>>>> ),
>>>> )
>>>>
>>>> Then, it does use the deprecated read.
>>>>
>>>> [image: image.png]
>>>> Cheers,
>>>> Jaehyeon
>>>>
>>>>
>>>> On Wed, 3 Jul 2024 at 06:44, Jaehyeon Kim <do...@gmail.com> wrote:
>>>>
>>>>> Hello,
>>>>>
>>>>> I'm using the Python SDK and the source can be found in
>>>>> https://github.com/jaehyeon-kim/beam-demos/blob/master/beam-pipelines/chapter2/top_k_words.py.
>>>>> I added the experiment flag but it doesn't seem to be applied. (Flink
>>>>> 1.18.1, Beam 2.57.0)
>>>>>
>>>>> [image: image.png]
>>>>>
>>>>> Can you please inform me how to disable `commitOffsetsInFinalize()` in
>>>>> Python?
>>>>>
>>>>> Cheers,
>>>>> Jaehyeon
>>>>>
>>>>> On Tue, 2 Jul 2024 at 21:33, Bartosz Zabłocki via user <
>>>>> user@beam.apache.org> wrote:
>>>>>
>>>>>> Hi Jaehyeon,
>>>>>> I don't see any changes that could lead to the behavior you noticed.
>>>>>>
>>>>>> KafkaIO read transform can be compiled in two modes
>>>>>> - ReadFromKafkaViaUnbounded or ReadFromKafkaViaSDF. The fist screenshot
>>>>>> shows ReadFromKafkaViaUnbounded and the second ReadFromKafkaViaSDF.
>>>>>>
>>>>>> Here's the logic from the KafkaIO.java
>>>>>> <https://github.com/apache/beam/blob/bf8afb8d58083f3372ab826eb83402b69db72e95/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java#L1465>
>>>>>> that chooses the implementation:
>>>>>>
>>>>>> ```
>>>>>> if (ExperimentalOptions.hasExperiment(
>>>>>> input.getPipeline().getOptions(),
>>>>>> "beam_fn_api_use_deprecated_read")
>>>>>> || ExperimentalOptions.hasExperiment(
>>>>>> input.getPipeline().getOptions(), "use_deprecated_read")
>>>>>> || ExperimentalOptions.hasExperiment(
>>>>>> input.getPipeline().getOptions(),
>>>>>> "use_unbounded_sdf_wrapper")
>>>>>> ||
>>>>>> compatibility.supportsOnly(KafkaIOReadImplementation.LEGACY)
>>>>>> || (compatibility.supports(KafkaIOReadImplementation.LEGACY)
>>>>>> &&
>>>>>> runnerPrefersLegacyRead(input.getPipeline().getOptions()))) {
>>>>>> return input.apply(new ReadFromKafkaViaUnbounded<>(this,
>>>>>> keyCoder, valueCoder));
>>>>>> }
>>>>>> return input.apply(new ReadFromKafkaViaSDF<>(this, keyCoder,
>>>>>> valueCoder));
>>>>>> ```
>>>>>>
>>>>>> In your case, the `--experiment=use_deprecated_read` should result in
>>>>>> the `ReadFromKafkaViaUnbounded`. Could you double-check if the flag is
>>>>>> enabled?
>>>>>> Could you share your pipeline definition? Are you using
>>>>>> `KafkaIO.read(). ... ` as the first step in the pipeline definition or do
>>>>>> you feed `ReadSourceDescriptors` to `KafkaIO.read()`?
>>>>>>
>>>>>> Btw, if you're going to use ReadFromKafkaViaSDF I recommend to *not*
>>>>>> using the `commitOffsetsInFinalize()` configuration parameter, it creates a
>>>>>> bottleneck for the SDF implementation. We're working on a fix.
>>>>>>
>>>>>> Cheers,
>>>>>> Bartosz
>>>>>>
>>>>>>
>>>>>> On Tue, Jul 2, 2024 at 12:28 PM Jaehyeon Kim <do...@gmail.com>
>>>>>> wrote:
>>>>>>
>>>>>>> Hello,
>>>>>>>
>>>>>>> A while ago, I deployed a beam pipeline on a Flink cluster (Flink
>>>>>>> 1.16.3) and the DAG only had two tasks.
>>>>>>>
>>>>>>> [image: image.png]
>>>>>>>
>>>>>>> When I deployed a similar pipeline on Flink 1.18.1, the DAG
>>>>>>> looked much more complicated.
>>>>>>>
>>>>>>> [image: image.png]
>>>>>>>
>>>>>>> In both the cases, I specified --experiment=use_deprecated_read.
>>>>>>>
>>>>>>> Is there a change in the KafkaIO configuration or I may have made a
>>>>>>> mistake to specifiy the pipeline argument?
>>>>>>>
>>>>>>> Cheers,
>>>>>>> Jaehyeon
>>>>>>>
>>>>>>
Re: KafkaIO no longer allows use_deprecated_read?
Posted by Jaehyeon Kim <do...@gmail.com>.
I don't think it is documented. Initially I noticed it in a post (
https://ndeepak.com/posts/2022-07-07-local-beam/), and it refers to a
GitHub issue (https://github.com/apache/beam/issues/20979). I followed both
the examples but didn't work until I added it directly to the java io
expansion service.
On Thu, 11 Jul 2024 at 01:46, Ahmet Altay via user <us...@beam.apache.org>
wrote:
> Thank you Jaehyeon.
>
> Was this (setting experiments via expansion_service) documented?
>
> On Wed, Jul 10, 2024 at 1:53 AM Bartosz Zabłocki via user <
> user@beam.apache.org> wrote:
>
>> Hi Jaehyeon,
>> thanks for the update and the instructions on how to do this! I wasn't
>> aware of the `expansion_service` argument.
>>
>> Cheers,
>> Bartosz
>>
>> On Wed, Jul 10, 2024 at 6:04 AM Jaehyeon Kim <do...@gmail.com> wrote:
>>
>>> Hello,
>>>
>>> Looks like I have to specify that option in the Kafka IO methods rather
>>> than specifying it as a pipeline argument. eg)
>>>
>>> >> kafka.ReadFromKafka(
>>> consumer_config={
>>> "bootstrap.servers": self.boostrap_servers,
>>> "auto.offset.reset": "earliest",
>>> # "enable.auto.commit": "true",
>>> "group.id": self.group_id,
>>> },
>>> topics=self.topics,
>>> timestamp_policy=kafka.ReadFromKafka.create_time_policy,
>>> expansion_service=kafka.default_io_expansion_service(
>>> ["--experiments=use_deprecated_read"]
>>> ),
>>> )
>>>
>>> Then, it does use the deprecated read.
>>>
>>> [image: image.png]
>>> Cheers,
>>> Jaehyeon
>>>
>>>
>>> On Wed, 3 Jul 2024 at 06:44, Jaehyeon Kim <do...@gmail.com> wrote:
>>>
>>>> Hello,
>>>>
>>>> I'm using the Python SDK and the source can be found in
>>>> https://github.com/jaehyeon-kim/beam-demos/blob/master/beam-pipelines/chapter2/top_k_words.py.
>>>> I added the experiment flag but it doesn't seem to be applied. (Flink
>>>> 1.18.1, Beam 2.57.0)
>>>>
>>>> [image: image.png]
>>>>
>>>> Can you please inform me how to disable `commitOffsetsInFinalize()` in
>>>> Python?
>>>>
>>>> Cheers,
>>>> Jaehyeon
>>>>
>>>> On Tue, 2 Jul 2024 at 21:33, Bartosz Zabłocki via user <
>>>> user@beam.apache.org> wrote:
>>>>
>>>>> Hi Jaehyeon,
>>>>> I don't see any changes that could lead to the behavior you noticed.
>>>>>
>>>>> KafkaIO read transform can be compiled in two modes
>>>>> - ReadFromKafkaViaUnbounded or ReadFromKafkaViaSDF. The fist screenshot
>>>>> shows ReadFromKafkaViaUnbounded and the second ReadFromKafkaViaSDF.
>>>>>
>>>>> Here's the logic from the KafkaIO.java
>>>>> <https://github.com/apache/beam/blob/bf8afb8d58083f3372ab826eb83402b69db72e95/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java#L1465>
>>>>> that chooses the implementation:
>>>>>
>>>>> ```
>>>>> if (ExperimentalOptions.hasExperiment(
>>>>> input.getPipeline().getOptions(),
>>>>> "beam_fn_api_use_deprecated_read")
>>>>> || ExperimentalOptions.hasExperiment(
>>>>> input.getPipeline().getOptions(), "use_deprecated_read")
>>>>> || ExperimentalOptions.hasExperiment(
>>>>> input.getPipeline().getOptions(),
>>>>> "use_unbounded_sdf_wrapper")
>>>>> ||
>>>>> compatibility.supportsOnly(KafkaIOReadImplementation.LEGACY)
>>>>> || (compatibility.supports(KafkaIOReadImplementation.LEGACY)
>>>>> &&
>>>>> runnerPrefersLegacyRead(input.getPipeline().getOptions()))) {
>>>>> return input.apply(new ReadFromKafkaViaUnbounded<>(this,
>>>>> keyCoder, valueCoder));
>>>>> }
>>>>> return input.apply(new ReadFromKafkaViaSDF<>(this, keyCoder,
>>>>> valueCoder));
>>>>> ```
>>>>>
>>>>> In your case, the `--experiment=use_deprecated_read` should result in
>>>>> the `ReadFromKafkaViaUnbounded`. Could you double-check if the flag is
>>>>> enabled?
>>>>> Could you share your pipeline definition? Are you using
>>>>> `KafkaIO.read(). ... ` as the first step in the pipeline definition or do
>>>>> you feed `ReadSourceDescriptors` to `KafkaIO.read()`?
>>>>>
>>>>> Btw, if you're going to use ReadFromKafkaViaSDF I recommend to *not*
>>>>> using the `commitOffsetsInFinalize()` configuration parameter, it creates a
>>>>> bottleneck for the SDF implementation. We're working on a fix.
>>>>>
>>>>> Cheers,
>>>>> Bartosz
>>>>>
>>>>>
>>>>> On Tue, Jul 2, 2024 at 12:28 PM Jaehyeon Kim <do...@gmail.com>
>>>>> wrote:
>>>>>
>>>>>> Hello,
>>>>>>
>>>>>> A while ago, I deployed a beam pipeline on a Flink cluster (Flink
>>>>>> 1.16.3) and the DAG only had two tasks.
>>>>>>
>>>>>> [image: image.png]
>>>>>>
>>>>>> When I deployed a similar pipeline on Flink 1.18.1, the DAG
>>>>>> looked much more complicated.
>>>>>>
>>>>>> [image: image.png]
>>>>>>
>>>>>> In both the cases, I specified --experiment=use_deprecated_read.
>>>>>>
>>>>>> Is there a change in the KafkaIO configuration or I may have made a
>>>>>> mistake to specifiy the pipeline argument?
>>>>>>
>>>>>> Cheers,
>>>>>> Jaehyeon
>>>>>>
>>>>>
Re: KafkaIO no longer allows use_deprecated_read?
Posted by Ahmet Altay via user <us...@beam.apache.org>.
Thank you Jaehyeon.
Was this (setting experiments via expansion_service) documented?
On Wed, Jul 10, 2024 at 1:53 AM Bartosz Zabłocki via user <
user@beam.apache.org> wrote:
> Hi Jaehyeon,
> thanks for the update and the instructions on how to do this! I wasn't
> aware of the `expansion_service` argument.
>
> Cheers,
> Bartosz
>
> On Wed, Jul 10, 2024 at 6:04 AM Jaehyeon Kim <do...@gmail.com> wrote:
>
>> Hello,
>>
>> Looks like I have to specify that option in the Kafka IO methods rather
>> than specifying it as a pipeline argument. eg)
>>
>> >> kafka.ReadFromKafka(
>> consumer_config={
>> "bootstrap.servers": self.boostrap_servers,
>> "auto.offset.reset": "earliest",
>> # "enable.auto.commit": "true",
>> "group.id": self.group_id,
>> },
>> topics=self.topics,
>> timestamp_policy=kafka.ReadFromKafka.create_time_policy,
>> expansion_service=kafka.default_io_expansion_service(
>> ["--experiments=use_deprecated_read"]
>> ),
>> )
>>
>> Then, it does use the deprecated read.
>>
>> [image: image.png]
>> Cheers,
>> Jaehyeon
>>
>>
>> On Wed, 3 Jul 2024 at 06:44, Jaehyeon Kim <do...@gmail.com> wrote:
>>
>>> Hello,
>>>
>>> I'm using the Python SDK and the source can be found in
>>> https://github.com/jaehyeon-kim/beam-demos/blob/master/beam-pipelines/chapter2/top_k_words.py.
>>> I added the experiment flag but it doesn't seem to be applied. (Flink
>>> 1.18.1, Beam 2.57.0)
>>>
>>> [image: image.png]
>>>
>>> Can you please inform me how to disable `commitOffsetsInFinalize()` in
>>> Python?
>>>
>>> Cheers,
>>> Jaehyeon
>>>
>>> On Tue, 2 Jul 2024 at 21:33, Bartosz Zabłocki via user <
>>> user@beam.apache.org> wrote:
>>>
>>>> Hi Jaehyeon,
>>>> I don't see any changes that could lead to the behavior you noticed.
>>>>
>>>> KafkaIO read transform can be compiled in two modes
>>>> - ReadFromKafkaViaUnbounded or ReadFromKafkaViaSDF. The fist screenshot
>>>> shows ReadFromKafkaViaUnbounded and the second ReadFromKafkaViaSDF.
>>>>
>>>> Here's the logic from the KafkaIO.java
>>>> <https://github.com/apache/beam/blob/bf8afb8d58083f3372ab826eb83402b69db72e95/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java#L1465>
>>>> that chooses the implementation:
>>>>
>>>> ```
>>>> if (ExperimentalOptions.hasExperiment(
>>>> input.getPipeline().getOptions(),
>>>> "beam_fn_api_use_deprecated_read")
>>>> || ExperimentalOptions.hasExperiment(
>>>> input.getPipeline().getOptions(), "use_deprecated_read")
>>>> || ExperimentalOptions.hasExperiment(
>>>> input.getPipeline().getOptions(),
>>>> "use_unbounded_sdf_wrapper")
>>>> ||
>>>> compatibility.supportsOnly(KafkaIOReadImplementation.LEGACY)
>>>> || (compatibility.supports(KafkaIOReadImplementation.LEGACY)
>>>> &&
>>>> runnerPrefersLegacyRead(input.getPipeline().getOptions()))) {
>>>> return input.apply(new ReadFromKafkaViaUnbounded<>(this,
>>>> keyCoder, valueCoder));
>>>> }
>>>> return input.apply(new ReadFromKafkaViaSDF<>(this, keyCoder,
>>>> valueCoder));
>>>> ```
>>>>
>>>> In your case, the `--experiment=use_deprecated_read` should result in
>>>> the `ReadFromKafkaViaUnbounded`. Could you double-check if the flag is
>>>> enabled?
>>>> Could you share your pipeline definition? Are you using
>>>> `KafkaIO.read(). ... ` as the first step in the pipeline definition or do
>>>> you feed `ReadSourceDescriptors` to `KafkaIO.read()`?
>>>>
>>>> Btw, if you're going to use ReadFromKafkaViaSDF I recommend to *not*
>>>> using the `commitOffsetsInFinalize()` configuration parameter, it creates a
>>>> bottleneck for the SDF implementation. We're working on a fix.
>>>>
>>>> Cheers,
>>>> Bartosz
>>>>
>>>>
>>>> On Tue, Jul 2, 2024 at 12:28 PM Jaehyeon Kim <do...@gmail.com> wrote:
>>>>
>>>>> Hello,
>>>>>
>>>>> A while ago, I deployed a beam pipeline on a Flink cluster (Flink
>>>>> 1.16.3) and the DAG only had two tasks.
>>>>>
>>>>> [image: image.png]
>>>>>
>>>>> When I deployed a similar pipeline on Flink 1.18.1, the DAG
>>>>> looked much more complicated.
>>>>>
>>>>> [image: image.png]
>>>>>
>>>>> In both the cases, I specified --experiment=use_deprecated_read.
>>>>>
>>>>> Is there a change in the KafkaIO configuration or I may have made a
>>>>> mistake to specifiy the pipeline argument?
>>>>>
>>>>> Cheers,
>>>>> Jaehyeon
>>>>>
>>>>
Re: KafkaIO no longer allows use_deprecated_read?
Posted by Bartosz Zabłocki via user <us...@beam.apache.org>.
Hi Jaehyeon,
thanks for the update and the instructions on how to do this! I wasn't
aware of the `expansion_service` argument.
Cheers,
Bartosz
On Wed, Jul 10, 2024 at 6:04 AM Jaehyeon Kim <do...@gmail.com> wrote:
> Hello,
>
> Looks like I have to specify that option in the Kafka IO methods rather
> than specifying it as a pipeline argument. eg)
>
> >> kafka.ReadFromKafka(
> consumer_config={
> "bootstrap.servers": self.boostrap_servers,
> "auto.offset.reset": "earliest",
> # "enable.auto.commit": "true",
> "group.id": self.group_id,
> },
> topics=self.topics,
> timestamp_policy=kafka.ReadFromKafka.create_time_policy,
> expansion_service=kafka.default_io_expansion_service(
> ["--experiments=use_deprecated_read"]
> ),
> )
>
> Then, it does use the deprecated read.
>
> [image: image.png]
> Cheers,
> Jaehyeon
>
>
> On Wed, 3 Jul 2024 at 06:44, Jaehyeon Kim <do...@gmail.com> wrote:
>
>> Hello,
>>
>> I'm using the Python SDK and the source can be found in
>> https://github.com/jaehyeon-kim/beam-demos/blob/master/beam-pipelines/chapter2/top_k_words.py.
>> I added the experiment flag but it doesn't seem to be applied. (Flink
>> 1.18.1, Beam 2.57.0)
>>
>> [image: image.png]
>>
>> Can you please inform me how to disable `commitOffsetsInFinalize()` in
>> Python?
>>
>> Cheers,
>> Jaehyeon
>>
>> On Tue, 2 Jul 2024 at 21:33, Bartosz Zabłocki via user <
>> user@beam.apache.org> wrote:
>>
>>> Hi Jaehyeon,
>>> I don't see any changes that could lead to the behavior you noticed.
>>>
>>> KafkaIO read transform can be compiled in two modes
>>> - ReadFromKafkaViaUnbounded or ReadFromKafkaViaSDF. The fist screenshot
>>> shows ReadFromKafkaViaUnbounded and the second ReadFromKafkaViaSDF.
>>>
>>> Here's the logic from the KafkaIO.java
>>> <https://github.com/apache/beam/blob/bf8afb8d58083f3372ab826eb83402b69db72e95/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java#L1465>
>>> that chooses the implementation:
>>>
>>> ```
>>> if (ExperimentalOptions.hasExperiment(
>>> input.getPipeline().getOptions(),
>>> "beam_fn_api_use_deprecated_read")
>>> || ExperimentalOptions.hasExperiment(
>>> input.getPipeline().getOptions(), "use_deprecated_read")
>>> || ExperimentalOptions.hasExperiment(
>>> input.getPipeline().getOptions(),
>>> "use_unbounded_sdf_wrapper")
>>> || compatibility.supportsOnly(KafkaIOReadImplementation.LEGACY)
>>> || (compatibility.supports(KafkaIOReadImplementation.LEGACY)
>>> &&
>>> runnerPrefersLegacyRead(input.getPipeline().getOptions()))) {
>>> return input.apply(new ReadFromKafkaViaUnbounded<>(this,
>>> keyCoder, valueCoder));
>>> }
>>> return input.apply(new ReadFromKafkaViaSDF<>(this, keyCoder,
>>> valueCoder));
>>> ```
>>>
>>> In your case, the `--experiment=use_deprecated_read` should result in
>>> the `ReadFromKafkaViaUnbounded`. Could you double-check if the flag is
>>> enabled?
>>> Could you share your pipeline definition? Are you using `KafkaIO.read().
>>> ... ` as the first step in the pipeline definition or do you
>>> feed `ReadSourceDescriptors` to `KafkaIO.read()`?
>>>
>>> Btw, if you're going to use ReadFromKafkaViaSDF I recommend to *not*
>>> using the `commitOffsetsInFinalize()` configuration parameter, it creates a
>>> bottleneck for the SDF implementation. We're working on a fix.
>>>
>>> Cheers,
>>> Bartosz
>>>
>>>
>>> On Tue, Jul 2, 2024 at 12:28 PM Jaehyeon Kim <do...@gmail.com> wrote:
>>>
>>>> Hello,
>>>>
>>>> A while ago, I deployed a beam pipeline on a Flink cluster (Flink
>>>> 1.16.3) and the DAG only had two tasks.
>>>>
>>>> [image: image.png]
>>>>
>>>> When I deployed a similar pipeline on Flink 1.18.1, the DAG looked much
>>>> more complicated.
>>>>
>>>> [image: image.png]
>>>>
>>>> In both the cases, I specified --experiment=use_deprecated_read.
>>>>
>>>> Is there a change in the KafkaIO configuration or I may have made a
>>>> mistake to specifiy the pipeline argument?
>>>>
>>>> Cheers,
>>>> Jaehyeon
>>>>
>>>
Re: KafkaIO no longer allows use_deprecated_read?
Posted by Jaehyeon Kim <do...@gmail.com>.
Hello,
Looks like I have to specify that option in the Kafka IO methods rather
than specifying it as a pipeline argument. eg)
>> kafka.ReadFromKafka(
consumer_config={
"bootstrap.servers": self.boostrap_servers,
"auto.offset.reset": "earliest",
# "enable.auto.commit": "true",
"group.id": self.group_id,
},
topics=self.topics,
timestamp_policy=kafka.ReadFromKafka.create_time_policy,
expansion_service=kafka.default_io_expansion_service(
["--experiments=use_deprecated_read"]
),
)
Then, it does use the deprecated read.
[image: image.png]
Cheers,
Jaehyeon
On Wed, 3 Jul 2024 at 06:44, Jaehyeon Kim <do...@gmail.com> wrote:
> Hello,
>
> I'm using the Python SDK and the source can be found in
> https://github.com/jaehyeon-kim/beam-demos/blob/master/beam-pipelines/chapter2/top_k_words.py.
> I added the experiment flag but it doesn't seem to be applied. (Flink
> 1.18.1, Beam 2.57.0)
>
> [image: image.png]
>
> Can you please inform me how to disable `commitOffsetsInFinalize()` in
> Python?
>
> Cheers,
> Jaehyeon
>
> On Tue, 2 Jul 2024 at 21:33, Bartosz Zabłocki via user <
> user@beam.apache.org> wrote:
>
>> Hi Jaehyeon,
>> I don't see any changes that could lead to the behavior you noticed.
>>
>> KafkaIO read transform can be compiled in two modes
>> - ReadFromKafkaViaUnbounded or ReadFromKafkaViaSDF. The fist screenshot
>> shows ReadFromKafkaViaUnbounded and the second ReadFromKafkaViaSDF.
>>
>> Here's the logic from the KafkaIO.java
>> <https://github.com/apache/beam/blob/bf8afb8d58083f3372ab826eb83402b69db72e95/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java#L1465>
>> that chooses the implementation:
>>
>> ```
>> if (ExperimentalOptions.hasExperiment(
>> input.getPipeline().getOptions(),
>> "beam_fn_api_use_deprecated_read")
>> || ExperimentalOptions.hasExperiment(
>> input.getPipeline().getOptions(), "use_deprecated_read")
>> || ExperimentalOptions.hasExperiment(
>> input.getPipeline().getOptions(),
>> "use_unbounded_sdf_wrapper")
>> || compatibility.supportsOnly(KafkaIOReadImplementation.LEGACY)
>> || (compatibility.supports(KafkaIOReadImplementation.LEGACY)
>> &&
>> runnerPrefersLegacyRead(input.getPipeline().getOptions()))) {
>> return input.apply(new ReadFromKafkaViaUnbounded<>(this,
>> keyCoder, valueCoder));
>> }
>> return input.apply(new ReadFromKafkaViaSDF<>(this, keyCoder,
>> valueCoder));
>> ```
>>
>> In your case, the `--experiment=use_deprecated_read` should result in the
>> `ReadFromKafkaViaUnbounded`. Could you double-check if the flag is enabled?
>> Could you share your pipeline definition? Are you using `KafkaIO.read().
>> ... ` as the first step in the pipeline definition or do you
>> feed `ReadSourceDescriptors` to `KafkaIO.read()`?
>>
>> Btw, if you're going to use ReadFromKafkaViaSDF I recommend to *not*
>> using the `commitOffsetsInFinalize()` configuration parameter, it creates a
>> bottleneck for the SDF implementation. We're working on a fix.
>>
>> Cheers,
>> Bartosz
>>
>>
>> On Tue, Jul 2, 2024 at 12:28 PM Jaehyeon Kim <do...@gmail.com> wrote:
>>
>>> Hello,
>>>
>>> A while ago, I deployed a beam pipeline on a Flink cluster (Flink
>>> 1.16.3) and the DAG only had two tasks.
>>>
>>> [image: image.png]
>>>
>>> When I deployed a similar pipeline on Flink 1.18.1, the DAG looked much
>>> more complicated.
>>>
>>> [image: image.png]
>>>
>>> In both the cases, I specified --experiment=use_deprecated_read.
>>>
>>> Is there a change in the KafkaIO configuration or I may have made a
>>> mistake to specifiy the pipeline argument?
>>>
>>> Cheers,
>>> Jaehyeon
>>>
>>
Re: KafkaIO no longer allows use_deprecated_read?
Posted by Jaehyeon Kim <do...@gmail.com>.
Hello,
I'm using the Python SDK and the source can be found in
https://github.com/jaehyeon-kim/beam-demos/blob/master/beam-pipelines/chapter2/top_k_words.py.
I added the experiment flag but it doesn't seem to be applied. (Flink
1.18.1, Beam 2.57.0)
[image: image.png]
Can you please inform me how to disable `commitOffsetsInFinalize()` in
Python?
Cheers,
Jaehyeon
On Tue, 2 Jul 2024 at 21:33, Bartosz Zabłocki via user <us...@beam.apache.org>
wrote:
> Hi Jaehyeon,
> I don't see any changes that could lead to the behavior you noticed.
>
> KafkaIO read transform can be compiled in two modes
> - ReadFromKafkaViaUnbounded or ReadFromKafkaViaSDF. The fist screenshot
> shows ReadFromKafkaViaUnbounded and the second ReadFromKafkaViaSDF.
>
> Here's the logic from the KafkaIO.java
> <https://github.com/apache/beam/blob/bf8afb8d58083f3372ab826eb83402b69db72e95/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java#L1465>
> that chooses the implementation:
>
> ```
> if (ExperimentalOptions.hasExperiment(
> input.getPipeline().getOptions(),
> "beam_fn_api_use_deprecated_read")
> || ExperimentalOptions.hasExperiment(
> input.getPipeline().getOptions(), "use_deprecated_read")
> || ExperimentalOptions.hasExperiment(
> input.getPipeline().getOptions(),
> "use_unbounded_sdf_wrapper")
> || compatibility.supportsOnly(KafkaIOReadImplementation.LEGACY)
> || (compatibility.supports(KafkaIOReadImplementation.LEGACY)
> &&
> runnerPrefersLegacyRead(input.getPipeline().getOptions()))) {
> return input.apply(new ReadFromKafkaViaUnbounded<>(this, keyCoder,
> valueCoder));
> }
> return input.apply(new ReadFromKafkaViaSDF<>(this, keyCoder,
> valueCoder));
> ```
>
> In your case, the `--experiment=use_deprecated_read` should result in the
> `ReadFromKafkaViaUnbounded`. Could you double-check if the flag is enabled?
> Could you share your pipeline definition? Are you using `KafkaIO.read().
> ... ` as the first step in the pipeline definition or do you
> feed `ReadSourceDescriptors` to `KafkaIO.read()`?
>
> Btw, if you're going to use ReadFromKafkaViaSDF I recommend to *not*
> using the `commitOffsetsInFinalize()` configuration parameter, it creates a
> bottleneck for the SDF implementation. We're working on a fix.
>
> Cheers,
> Bartosz
>
>
> On Tue, Jul 2, 2024 at 12:28 PM Jaehyeon Kim <do...@gmail.com> wrote:
>
>> Hello,
>>
>> A while ago, I deployed a beam pipeline on a Flink cluster (Flink 1.16.3)
>> and the DAG only had two tasks.
>>
>> [image: image.png]
>>
>> When I deployed a similar pipeline on Flink 1.18.1, the DAG looked much
>> more complicated.
>>
>> [image: image.png]
>>
>> In both the cases, I specified --experiment=use_deprecated_read.
>>
>> Is there a change in the KafkaIO configuration or I may have made a
>> mistake to specifiy the pipeline argument?
>>
>> Cheers,
>> Jaehyeon
>>
>
Re: KafkaIO no longer allows use_deprecated_read?
Posted by Bartosz Zabłocki via user <us...@beam.apache.org>.
Hi Jaehyeon,
I don't see any changes that could lead to the behavior you noticed.
KafkaIO read transform can be compiled in two modes
- ReadFromKafkaViaUnbounded or ReadFromKafkaViaSDF. The fist screenshot
shows ReadFromKafkaViaUnbounded and the second ReadFromKafkaViaSDF.
Here's the logic from the KafkaIO.java
<https://github.com/apache/beam/blob/bf8afb8d58083f3372ab826eb83402b69db72e95/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java#L1465>
that chooses the implementation:
```
if (ExperimentalOptions.hasExperiment(
input.getPipeline().getOptions(),
"beam_fn_api_use_deprecated_read")
|| ExperimentalOptions.hasExperiment(
input.getPipeline().getOptions(), "use_deprecated_read")
|| ExperimentalOptions.hasExperiment(
input.getPipeline().getOptions(), "use_unbounded_sdf_wrapper")
|| compatibility.supportsOnly(KafkaIOReadImplementation.LEGACY)
|| (compatibility.supports(KafkaIOReadImplementation.LEGACY)
&&
runnerPrefersLegacyRead(input.getPipeline().getOptions()))) {
return input.apply(new ReadFromKafkaViaUnbounded<>(this, keyCoder,
valueCoder));
}
return input.apply(new ReadFromKafkaViaSDF<>(this, keyCoder,
valueCoder));
```
In your case, the `--experiment=use_deprecated_read` should result in the
`ReadFromKafkaViaUnbounded`. Could you double-check if the flag is enabled?
Could you share your pipeline definition? Are you using `KafkaIO.read().
... ` as the first step in the pipeline definition or do you
feed `ReadSourceDescriptors` to `KafkaIO.read()`?
Btw, if you're going to use ReadFromKafkaViaSDF I recommend to *not* using
the `commitOffsetsInFinalize()` configuration parameter, it creates a
bottleneck for the SDF implementation. We're working on a fix.
Cheers,
Bartosz
On Tue, Jul 2, 2024 at 12:28 PM Jaehyeon Kim <do...@gmail.com> wrote:
> Hello,
>
> A while ago, I deployed a beam pipeline on a Flink cluster (Flink 1.16.3)
> and the DAG only had two tasks.
>
> [image: image.png]
>
> When I deployed a similar pipeline on Flink 1.18.1, the DAG looked much
> more complicated.
>
> [image: image.png]
>
> In both the cases, I specified --experiment=use_deprecated_read.
>
> Is there a change in the KafkaIO configuration or I may have made a
> mistake to specifiy the pipeline argument?
>
> Cheers,
> Jaehyeon
>