You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@beam.apache.org by Lina Mårtensson via user <us...@beam.apache.org> on 2026/03/27 00:46:51 UTC
Dataflow jobs get really really slow then die
Hi beam-users,
I'm not sure if this is the right place for a Dataflow question - I think
this is where the issue appears.
I've been trying to run some large backfills since last night and they run
well, then get to almost 100 % on the main fused part of the pipeline
ending with a GroupByKey, and the job gets stuck. Most of my jobs haven't
finished - some of the small-ish-er ones have *eventually* finished after
several hours of being stuck at 97-99 %.
We ran this job recently with no issues, and with much more data in one go.
To try to isolate the issue, I ran the current data with the older image
where the job succeeded, and I also ran (a subset of) the older data with
the current image. Both of these jobs also got stuck, so that leads me to
believe it's a Dataflow issue rather than an issue on our end.
All our jobs have --dataflow_service_options = ['enable_prime'] set, which
I know is necessary for larger jobs where GroupByKey is used.
Is there anything going on with Dataflow right now that might cause this
issue?
Is this even the right place to ask?
Thanks!
-Lina
Re: Dataflow jobs get really really slow then die
Posted by Jonathan Hope <jo...@gmail.com>.
At least with go there is a subtle memory leak if you iterate over one key
with billions of things under it. I has to add synthetic sharding to a
large batch job to get it to work.
On Thu, Mar 26, 2026 at 5:55 PM Lina Mårtensson via user <
user@beam.apache.org> wrote:
> So sorry, I forgot to include the log messages!
>
> Screenshot attached for the messages that come in constantly during the
> stuck phase.
>
> Here's an example output from when a job finally crashes:
> Operation ongoing in bundle process_bundle-1864764743973176211-1 for
> PTransform{name=Read/Read
> files/Read/SDFBoundedSourceReader/ParDo(SDFBoundedSourceDoFn)/ProcessElementAndRestrictionWithSizing-ptransform-56,
> state=process-msecs} for at least 1986.44 seconds without outputting or
> completing.
> Current Traceback:
> File "/usr/local/lib/python3.10/threading.py", line 973, in _bootstrap
> self._bootstrap_inner()
>
> File "/usr/local/lib/python3.10/threading.py", line 1016, in
> _bootstrap_inner
> self.run()
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/utils/thread_pool_executor.py",
> line 53, in run
> self._work_item.run()
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/utils/thread_pool_executor.py",
> line 37, in run
> self._future.set_result(self._fn(*self._fn_args, **self._fn_kwargs))
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/runners/worker/sdk_worker.py",
> line 385, in task
> self._execute(
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/runners/worker/sdk_worker.py",
> line 311, in _execute
> response = task()
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/runners/worker/sdk_worker.py",
> line 386, in <lambda>
> lambda: self.create_worker().do_instruction(request), request)
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/runners/worker/sdk_worker.py",
> line 656, in do_instruction
> return getattr(self, request_type)(
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/runners/worker/sdk_worker.py",
> line 694, in process_bundle
> bundle_processor.process_bundle(instruction_id))
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/runners/worker/bundle_processor.py",
> line 1119, in process_bundle
> input_op_by_transform_id[element.transform_id].process_encoded(
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/runners/worker/bundle_processor.py",
> line 237, in process_encoded
> self.output(decoded_value)
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/io/textio.py",
> line 196, in read_records
> self._process_header(file_to_read, read_buffer))
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/io/textio.py",
> line 259, in _process_header
> position = self._skip_lines(
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/io/textio.py",
> line 360, in _skip_lines
> _, num_bytes_to_next_record = self._read_record(file_to_read,
> read_buffer)
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/io/textio.py",
> line 379, in _read_record
> sep_bounds = self._find_separator_bounds(file_to_read, read_buffer)
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/io/textio.py",
> line 300, in _find_separator_bounds
> if not self._try_to_ensure_num_bytes_in_buffer(
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/io/textio.py",
> line 350, in _try_to_ensure_num_bytes_in_buffer
> read_buffer.data += read_data
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/io/textio.py",
> line 83, in data
> @data.setter
>
> [image: Screenshot 2026-03-26 at 5.52.25 PM.png]
>
> On Thu, Mar 26, 2026 at 5:46 PM Lina Mårtensson <li...@camus.energy> wrote:
>
>> Hi beam-users,
>>
>> I'm not sure if this is the right place for a Dataflow question - I think
>> this is where the issue appears.
>>
>> I've been trying to run some large backfills since last night and they
>> run well, then get to almost 100 % on the main fused part of the pipeline
>> ending with a GroupByKey, and the job gets stuck. Most of my jobs
>> haven't finished - some of the small-ish-er ones have *eventually* finished
>> after several hours of being stuck at 97-99 %.
>>
>> We ran this job recently with no issues, and with much more data in one
>> go.
>> To try to isolate the issue, I ran the current data with the older image
>> where the job succeeded, and I also ran (a subset of) the older data with
>> the current image. Both of these jobs also got stuck, so that leads me to
>> believe it's a Dataflow issue rather than an issue on our end.
>>
>> All our jobs have --dataflow_service_options = ['enable_prime'] set,
>> which I know is necessary for larger jobs where GroupByKey is used.
>>
>> Is there anything going on with Dataflow right now that might cause this
>> issue?
>> Is this even the right place to ask?
>>
>> Thanks!
>> -Lina
>>
>
Re: Dataflow jobs get really really slow then die
Posted by Shunping Huang via user <us...@beam.apache.org>.
For dataflow related issues, please open a customer support ticket. The
product team will investigate that further.
You can find more information here:
https://docs.cloud.google.com/dataflow/docs/support/getting-support
On Thu, Mar 26, 2026 at 8:57 PM Lina Mårtensson via user <
user@beam.apache.org> wrote:
> So sorry, I forgot to include the log messages!
>
> Screenshot attached for the messages that come in constantly during the
> stuck phase.
>
> Here's an example output from when a job finally crashes:
> Operation ongoing in bundle process_bundle-1864764743973176211-1 for
> PTransform{name=Read/Read
> files/Read/SDFBoundedSourceReader/ParDo(SDFBoundedSourceDoFn)/ProcessElementAndRestrictionWithSizing-ptransform-56,
> state=process-msecs} for at least 1986.44 seconds without outputting or
> completing.
> Current Traceback:
> File "/usr/local/lib/python3.10/threading.py", line 973, in _bootstrap
> self._bootstrap_inner()
>
> File "/usr/local/lib/python3.10/threading.py", line 1016, in
> _bootstrap_inner
> self.run()
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/utils/thread_pool_executor.py",
> line 53, in run
> self._work_item.run()
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/utils/thread_pool_executor.py",
> line 37, in run
> self._future.set_result(self._fn(*self._fn_args, **self._fn_kwargs))
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/runners/worker/sdk_worker.py",
> line 385, in task
> self._execute(
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/runners/worker/sdk_worker.py",
> line 311, in _execute
> response = task()
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/runners/worker/sdk_worker.py",
> line 386, in <lambda>
> lambda: self.create_worker().do_instruction(request), request)
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/runners/worker/sdk_worker.py",
> line 656, in do_instruction
> return getattr(self, request_type)(
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/runners/worker/sdk_worker.py",
> line 694, in process_bundle
> bundle_processor.process_bundle(instruction_id))
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/runners/worker/bundle_processor.py",
> line 1119, in process_bundle
> input_op_by_transform_id[element.transform_id].process_encoded(
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/runners/worker/bundle_processor.py",
> line 237, in process_encoded
> self.output(decoded_value)
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/io/textio.py",
> line 196, in read_records
> self._process_header(file_to_read, read_buffer))
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/io/textio.py",
> line 259, in _process_header
> position = self._skip_lines(
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/io/textio.py",
> line 360, in _skip_lines
> _, num_bytes_to_next_record = self._read_record(file_to_read,
> read_buffer)
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/io/textio.py",
> line 379, in _read_record
> sep_bounds = self._find_separator_bounds(file_to_read, read_buffer)
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/io/textio.py",
> line 300, in _find_separator_bounds
> if not self._try_to_ensure_num_bytes_in_buffer(
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/io/textio.py",
> line 350, in _try_to_ensure_num_bytes_in_buffer
> read_buffer.data += read_data
>
> File
> "/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/io/textio.py",
> line 83, in data
> @data.setter
>
> [image: Screenshot 2026-03-26 at 5.52.25 PM.png]
>
> On Thu, Mar 26, 2026 at 5:46 PM Lina Mårtensson <li...@camus.energy> wrote:
>
>> Hi beam-users,
>>
>> I'm not sure if this is the right place for a Dataflow question - I think
>> this is where the issue appears.
>>
>> I've been trying to run some large backfills since last night and they
>> run well, then get to almost 100 % on the main fused part of the pipeline
>> ending with a GroupByKey, and the job gets stuck. Most of my jobs
>> haven't finished - some of the small-ish-er ones have *eventually* finished
>> after several hours of being stuck at 97-99 %.
>>
>> We ran this job recently with no issues, and with much more data in one
>> go.
>> To try to isolate the issue, I ran the current data with the older image
>> where the job succeeded, and I also ran (a subset of) the older data with
>> the current image. Both of these jobs also got stuck, so that leads me to
>> believe it's a Dataflow issue rather than an issue on our end.
>>
>> All our jobs have --dataflow_service_options = ['enable_prime'] set,
>> which I know is necessary for larger jobs where GroupByKey is used.
>>
>> Is there anything going on with Dataflow right now that might cause this
>> issue?
>> Is this even the right place to ask?
>>
>> Thanks!
>> -Lina
>>
>
Re: Dataflow jobs get really really slow then die
Posted by Lina Mårtensson via user <us...@beam.apache.org>.
So sorry, I forgot to include the log messages!
Screenshot attached for the messages that come in constantly during the
stuck phase.
Here's an example output from when a job finally crashes:
Operation ongoing in bundle process_bundle-1864764743973176211-1 for
PTransform{name=Read/Read
files/Read/SDFBoundedSourceReader/ParDo(SDFBoundedSourceDoFn)/ProcessElementAndRestrictionWithSizing-ptransform-56,
state=process-msecs} for at least 1986.44 seconds without outputting or
completing.
Current Traceback:
File "/usr/local/lib/python3.10/threading.py", line 973, in _bootstrap
self._bootstrap_inner()
File "/usr/local/lib/python3.10/threading.py", line 1016, in
_bootstrap_inner
self.run()
File
"/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/utils/thread_pool_executor.py",
line 53, in run
self._work_item.run()
File
"/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/utils/thread_pool_executor.py",
line 37, in run
self._future.set_result(self._fn(*self._fn_args, **self._fn_kwargs))
File
"/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/runners/worker/sdk_worker.py",
line 385, in task
self._execute(
File
"/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/runners/worker/sdk_worker.py",
line 311, in _execute
response = task()
File
"/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/runners/worker/sdk_worker.py",
line 386, in <lambda>
lambda: self.create_worker().do_instruction(request), request)
File
"/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/runners/worker/sdk_worker.py",
line 656, in do_instruction
return getattr(self, request_type)(
File
"/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/runners/worker/sdk_worker.py",
line 694, in process_bundle
bundle_processor.process_bundle(instruction_id))
File
"/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/runners/worker/bundle_processor.py",
line 1119, in process_bundle
input_op_by_transform_id[element.transform_id].process_encoded(
File
"/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/runners/worker/bundle_processor.py",
line 237, in process_encoded
self.output(decoded_value)
File
"/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/io/textio.py",
line 196, in read_records
self._process_header(file_to_read, read_buffer))
File
"/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/io/textio.py",
line 259, in _process_header
position = self._skip_lines(
File
"/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/io/textio.py",
line 360, in _skip_lines
_, num_bytes_to_next_record = self._read_record(file_to_read,
read_buffer)
File
"/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/io/textio.py",
line 379, in _read_record
sep_bounds = self._find_separator_bounds(file_to_read, read_buffer)
File
"/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/io/textio.py",
line 300, in _find_separator_bounds
if not self._try_to_ensure_num_bytes_in_buffer(
File
"/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/io/textio.py",
line 350, in _try_to_ensure_num_bytes_in_buffer
read_buffer.data += read_data
File
"/app/ingestion/bigtable/ami/lke_beam_ami_parser_binary.runfiles/rules_python~~pip~common_deps_310_apache_beam/site-packages/apache_beam/io/textio.py",
line 83, in data
@data.setter
[image: Screenshot 2026-03-26 at 5.52.25 PM.png]
On Thu, Mar 26, 2026 at 5:46 PM Lina Mårtensson <li...@camus.energy> wrote:
> Hi beam-users,
>
> I'm not sure if this is the right place for a Dataflow question - I think
> this is where the issue appears.
>
> I've been trying to run some large backfills since last night and they run
> well, then get to almost 100 % on the main fused part of the pipeline
> ending with a GroupByKey, and the job gets stuck. Most of my jobs haven't
> finished - some of the small-ish-er ones have *eventually* finished after
> several hours of being stuck at 97-99 %.
>
> We ran this job recently with no issues, and with much more data in one go.
> To try to isolate the issue, I ran the current data with the older image
> where the job succeeded, and I also ran (a subset of) the older data with
> the current image. Both of these jobs also got stuck, so that leads me to
> believe it's a Dataflow issue rather than an issue on our end.
>
> All our jobs have --dataflow_service_options = ['enable_prime'] set,
> which I know is necessary for larger jobs where GroupByKey is used.
>
> Is there anything going on with Dataflow right now that might cause this
> issue?
> Is this even the right place to ask?
>
> Thanks!
> -Lina
>