[Bug]: [Python] DirectRunner Prism fallback to BundleBasedDirectRunner fails with NotImplementedError on batched DoFns (WriteToParquet)

オープン
  • 何か問題がありますか？
#41,078 コメント 0 件 リアクション 0 件 担当者 0 名 GitHub で見る

メンテナーはふだん 1 日以内に返信

まだ誰も着手していません。

評価

難易度
4/5
見積もり時間
3〜5日
初心者へのやさしさ
48/100
issue の種類
バグ
明瞭さ
おおむね明確
活発さ
活発
技術スタック
python
領域
backend, data-engineering

調査の方向性

Start in sdks/python/apache_beam/runners/direct/direct_runner.py, especially the SwitchingDirectRunner fallback around lines 141–146 and 204–230; compare its behavior with FnApiRunner for process_batch transforms. Reproduce with the provided WriteToParquet pipeline, then check relevant direct runner tests. Done means the fallback runs batched DoFns successfully without breaking the existing runner paths.

索引モデルが issue の本文から書いたものです。

説明

awaiting triage bug P2 python
What happened?

When running a Python or Beam YAML pipeline with the default DirectRunner (SwitchingDirectRunner), if PrismRunner is bypassed (for example when the pipeline contains an external cross-language transform per direct_runner.py:141) or cannot start (offline environment / restricted container where the Prism binary cannot be downloaded or executed), SwitchingDirectRunner falls back to BundleBasedDirectRunner.

If the pipeline includes WriteToParquet (or any transform using DoFn.process_batch), execution then fails with NotImplementedError, even though the in-process FnApiRunner executes the exact same pipeline cleanly.

Minimal reproducer (forcing the non-Prism fallback path):

import tempfile, apache_beam as beam, pyarrow as pa

schema = pa.schema([('id', pa.int64()), ('name', pa.string())])
with tempfile.TemporaryDirectory() as d:
  with beam.Pipeline(runner='BundleBasedDirectRunner') as p:
    (p
     | beam.Create([{'id': 1, 'name': 'Ada'}])
     | beam.io.WriteToParquet(f'{d}/out', schema=schema))

Cause

In sdks/python/apache_beam/runners/direct/direct_runner.py (lines 141–146 and 204–230), SwitchingDirectRunner unconditionally falls back to BundleBasedDirectRunner(), which does not implement process_batch for batched PyArrow DoFns (FnApiRunner does). Checking for process_batch transforms (or routing batch pipelines without TestStream/unbounded state to FnApiRunner when Prism is unavailable) avoids the failure.

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: Hazelcast Jet Runner
  • Component: Google Cloud Dataflow Runner
主要言語
Java
スター
8.7k
フォーク
4.8k
平均マージ
2日 12時間
マージ済み PR（30日）
371

環境構築

はじめの一歩

  1. issue を最後まで読み、次にプロジェクトのコントリビューションガイドを読みます。
  2. 着手することを issue にコメントします — 二人が同じ作業をするのを防げます。
  3. リポジトリをフォークし、ブランチを切って変更します。
  4. issue 番号を参照したプルリクエストを送ります。

apache/beam のほかの issue

apache/beam の issue をすべて見る

似ている issue

Java の issue をもっと見る

新しい issue をメールで受け取る

初心者向けの GitHub issue を短くまとめたダイジェスト。