Hacktoberfest 2026: những issue maintainer đã đánh dấu cho tháng Mười, đang mở và phù hợp người mới. Xem issue Hacktoberfest

[Bug]: DaskRunner `DaskBagWindowedIterator` materializes entire dataset into memory causing OOM

Đang mở
  • Có gì không ổn?
#40,282 0 bình luận 0 reaction 0 người được giao Xem trên GitHub

Maintainer thường phản hồi trong vòng 1 ngày

Chưa có ai nhận issue này.

  • #40283 của @vishalmore90 — đã đóng, không merge

Đánh giá

Độ khó
4/5
Thời gian dự kiến
3-5 ngày
Mức phù hợp với người mới
56/100
Loại issue
Lỗi
Độ rõ ràng
Khá rõ ràng
Mức độ hoạt động
Sôi nổi
Công nghệ
python
Lĩnh vực
data-engineering, distributed-systems

Hướng nghiên cứu

Start in sdks/python/apache_beam/runners/dask/transform_evaluator.py at DaskBagWindowedIterator.iter and review the FIXME around list(self.bag). Investigate the partition-based and delayed-result approaches described in the issue. Done means side-input results are yielded incrementally without materializing the full Dask Bag in client memory, avoiding the reported OOM behavior.

Do mô hình lập chỉ mục viết ra từ nội dung của issue.

Mô tả

bug P2 python
What happened?

In the experimental Python Dask Runner, the DaskBagWindowedIterator (which handles iterators for apache_beam.transforms.sideinputs.SideInputMap) iterates over a Dask Bag by wrapping it in a Python list().

Calling list(self.bag) implicitly triggers a full compute() on the Dask dataset. This blocking operation materializes the entirety of the side input data into the client's local memory. For large side inputs, this completely bypasses Dask's distributed memory management and results in an Out-Of-Memory (OOM) crash, effectively bottlenecking the scalability of pipelines running on Dask.

The code currently includes an explicit FIXME acknowledging this proof-of-concept behavior, but it remains a silent, critical scalability flaw.

Code Pointers / Steps to Reproduce

The issue is located in sdks/python/apache_beam/runners/dask/transform_evaluator.py within the __iter__ method of the DaskBagWindowedIterator class (lines 91-96):

class DaskBagWindowedIterator:
  """Iterator for `apache_beam.transforms.sideinputs.SideInputMap`"""

  bag: db.Bag
  window_fn: WindowFn

  def __iter__(self):
    # FIXME(cisaacstern): list() is likely inefficient, since it presumably
    # materializes the full result before iterating over it. doing this for
    # now as a proof-of-concept. can we can generate results incrementally?
    for result in list(self.bag):
      yield get_windowed_value(result, self.window_fn)

Impact

Any Apache Beam pipeline using DaskRunner that relies on substantial side inputs will crash with OOM errors as soon as the side input data surpasses the available local RAM on the node where the iterator is evaluated. This severely limits the DaskRunner's ability to process real-world distributed datasets and creates a harsh scalability ceiling.

Proposed Solution

The evaluation should generate results incrementally rather than performing a monolithic evaluation. Potential approaches:

  1. Partition-based iteration: Utilize Dask's .map_partitions or .to_delayed() to fetch and yield the underlying data partition-by-partition.
  2. Generators: Instead of eager computation via list(), retrieve delayed results asynchronously and yield them to allow the Python garbage collector to free memory between partition iterations.
Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

Issue Components
  • Component: Python SDK
  • Component: Java SDK
  • Component: Go SDK
  • Component: Typescript SDK
  • Component: IO connector
  • Component: Beam YAML
  • Component: Beam examples
  • Component: Beam playground
  • Component: Beam katas
  • Component: Website
  • Component: Infrastructure
  • Component: Spark Runner
  • Component: Flink Runner
  • Component: Prism Runner
  • Component: Twister2 Runner
  • Component: Hazelcast Jet Runner
  • Component: Google Cloud Dataflow Runner
Ngôn ngữ chính
Java
Star
8.7k
Fork
4.7k
Merge trung bình
2 ngày 14 giờ
Pull request đã merge (30 ngày)
300

Chuẩn bị môi trường

Bắt đầu từ đâu

  1. Đọc hết issue, rồi đọc hướng dẫn đóng góp của dự án.
  2. Bình luận trên issue rằng bạn sẽ nhận — tránh hai người làm cùng một việc.
  3. Fork repository và làm thay đổi trên một nhánh.
  4. Mở pull request có tham chiếu số hiệu của issue.

Issue khác của apache/beam

Tất cả issue của apache/beam

Issue tương tự

Thêm issue về Java

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.