[Bug]: RuntimeError: No dataflow job was found when running the python file.

[Ox0400]

What happened?

The main points of my issue are as follows:

When submitting a DataFlowPythonOP task to run beam pipeline, the script is executed using subprocess. However, apache_beam.runners.dataflow.internal.apiclient does not adjust the default log level, and logging.info() is used to print the Job ID and URL. As a result, create_python_job fails to match the Job ID, causing the task to fail.

Although it’s possible to forcibly set the log level in the script using:

logging.getLogger('apache_beam.runners.dataflow.internal.apiclient').setLevel(logging.INFO)  

this is not a very user-friendly solution.

Additional Context:
Vertex Pipeline needs to match the submitted Job URL from subprocess when running DataFlowPythonOP to call create_python_job. However, since the module does not change the default log level, the Job URL is not displayed, leading to a RuntimeError.

# run_pipeline.py
from google_cloud_pipeline_components.v1.dataflow import DataflowPythonJobOp
...

@dsl.pipeline(name='test dataflow')
def test_dataflow():
    dataflow_task = DataflowPythonJobOp(
        project=project,
        location=region,
        python_module_path=dataflow_clean_local_path,
        requirements_file_path=requirements_file_path,
        temp_location=temp_location,
        args=[
            "--project", project,
            "--region", region,
            "--temp_location", temp_location,
            "--job_name", f"dataflow-clean-{time.strftime('%Y%m%d-%H%M%S', time.gmtime())}",
            "--save_main_session",
            "--runner", "DataflowRunner",
        ],
    )

...

# data_clean.py
p = beam.Pipeline(options=options)
p | .....
result = p.run()
result.wait_until_finish()
return result

PR: #34952

Let me know if you'd like any refinements!

Issue Priority

Priority: 1 (data loss / total loss of function)

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: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[liferoad]

Can we set the log level inside DataflowPythonJobOp?

[shunping]

Can we set the log level inside DataflowPythonJobOp?

Probably not, since the user pipeline (i.e. data_clean.py) is executed through a subprocess. Setting the log level in DataflowPythonJobOp has no effect on that.

Could we just set the logging level in data_clean.py? It feels inappropriate to use warning level (as proposed in #34952) for the job id.

[Ox0400]

You're absolutely right—it's impossible to set the log level in DataflowPythonJobOp. The only workaround is to configure it in data_clean.py, which is what I did by adding:

logging.getLogger('apache_beam.runners.dataflow.internal.apiclient').setLevel(logging.INFO)

However, this solution is far from ideal:

Manual Overhead – Every script must explicitly set this, adding unnecessary boilerplate.
Undocumented Pitfall – There’s zero documentation warning about this requirement. Debugging this trivial issue took me days, only to discover it was caused by insufficient log levels preventing the parent process from retrieving the job ID.
Key Flaws in Current Design:

Poor Defaults – Logs critical for debugging (e.g., job IDs) should always be visible by default, not hidden behind manual configuration.
Fragile UX – Even if documented, users might overlook this step, leading to avoidable failures. Adding a "warning" for a single line of code feels like a band-aid rather than a proper fix.
Suggested Improvement:
The framework should auto-enable INFO logs for critical components (like job tracking) or fail fast with clear errors if log filtering blocks essential data. Silent failures due to log levels are a developer nightmare.

Certainly, there are two other options:

• 1, Use print() directly.
• 2, Set the log level of the current file's LOGGER to INFO.

[liferoad]

data_clean.py returns result, which should contian the job information like https://github.com/apache/beam/blob/master/sdks/python/apache_beam/runners/dataflow/dataflow_runner.py#L717. Relying on the logs is not good.

[Ox0400]

@liferoad You can review this source code dataflow_python_job_remote_runner.py. It's running on remote(google_cloud_pipeline_components) and using subprocess.

https://github.com/kubeflow/pipelines/blob/master/components/google-cloud/google_cloud_pipeline_components/container/v1/dataflow/dataflow_python_job_remote_runner.py

  for line in sub_process.read_lines():
    logging.info('DataflowRunner output: %s', line)
    job_id, location = extract_job_id_and_location(line)
    if job_id:
      logging.info('Found job id %s and location %s.', job_id, location)
      ...
      break
  if not job_id:
    raise RuntimeError(
        'No dataflow job was found when running the python file.'
    )

https://github.com/kubeflow/pipelines/blob/master/components/google-cloud/google_cloud_pipeline_components/container/v1/dataflow/dataflow_python_job_remote_runner.py#L33

https://github.com/kubeflow/pipelines/blob/master/components/google-cloud/google_cloud_pipeline_components/container/v1/dataflow/dataflow_python_job_remote_runner.py#L139

https://github.com/kubeflow/pipelines/blob/master/components/google-cloud/google_cloud_pipeline_components/container/v1/dataflow/dataflow_python_job_remote_runner.py#L106