[Python] Unable to trigger dynamically created pipelines in airflow DAG

I have a DAG file to load data from mongodb to bigquery. I have tested my apache beam pipeline individually but when i am trying to trigger it dynamically to do parallel processing of all collections in mongodb by creating BeamRunPythonPipelineOperator using a PythonOperator its not working. I am not getting any error but its not triggering jobs also.

from __future__ import annotations
import os

os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = "/gcp-creds/gcp_service_account.json"

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.apache.beam.operators.beam import BeamRunPythonPipelineOperator
from airflow.operators.empty import EmptyOperator

import pendulum
import logging

logging.getLogger().setLevel(logging.DEBUG)

default_args = {
'owner': 'airflow',
'start_date': datetime(year=2025, month=4, day=1, hour=00, minute=00, tzinfo=tz),
'retries': 1,
'retry_delay': timedelta(minutes=1)
}

beam_pipeline_file = "/opt/airflow/pipelines/local/mongo_bigquery.py"
beam_runner = "DataflowRunner"
mongo_url = "mongodb://localhost:27017/test"


def get_mongo_collections(**kwargs):
"""List of mongodb collections to load."""
return ["users", "orders"]


def create_beam_tasks(**kwargs):
"""Dynamically create Beam tasks for each collection."""
ti = kwargs['ti']
dag = kwargs['dag']
collections = ti.xcom_pull(task_ids='Get_MongoDB_collections')
beam_tasks = []
if collections:
for collection in collections:
logging.info(f"Creating Beam task for collection: {collection}")
beam_task = BeamRunPythonPipelineOperator(
task_id=f"mongodb_beam_pipeline_{collection}",
py_file=beam_pipeline_file,
runner=beam_runner,
py_options=[],
pipeline_options={
"mongo_uri": mongo_url,
"mongo_db": "test",
"mongo_collection": collection,
"bigquery_table": "project_id.testing.users", # later dynamically
"temp_location": "gs://tms-test1/beam/tmp/",
"setup_file": "/opt/airflow/pipelines/local/setup.py",
},
dataflow_config={
"project_id": "project_id",
"location": "us-central",
},
py_requirements=[
'apache-beam[gcp]==2.64.0',
'setuptools',
'pymongo',
],
py_system_site_packages=False,
dag=dag
)
beam_tasks.append(beam_task)
# return beam_tasks


with DAG(
dag_id='tmp_dag1',
default_args=default_args,
max_active_runs=1,
schedule_interval='@once',
description='A DAG to run Beam pipelines for MongoDB to BigQuery',
) as dag:
start_task = EmptyOperator(
task_id="start_task",
)

mongo_collections_task = PythonOperator(
task_id='Get_MongoDB_collections',
python_callable=get_mongo_collections,
do_xcom_push=True,
)

create_beam_tasks_op = PythonOperator(
task_id='Create_Beam_Tasks',
python_callable=create_beam_tasks,
provide_context=True, # We need access to Task Instance to pull XCom
)

end_task = EmptyOperator(
task_id="end_task",
)

start_task >> mongo_collections_task >> create_beam_tasks_op >> end_task

Continue reading...