Composer气流动态DAG中的"Dag似乎缺失“错误

Question

我有一个动态气流DAG在创建，列出在网络服务器和运行(回填)没有错误。然而，也存在一些问题：

1. 在web url中单击DAG时，它会说"DAG似乎丢失了“。
2. 无法看到图形视图/树视图显示上面的错误
3. 不能手动触发DAG以显示上面的错误

试图修复这一对days...any提示将是有帮助的。谢谢!

from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
from airflow.contrib.operators.gcs_to_bq import GoogleCloudStorageToBigQueryOperator
from google.cloud import storage
from airflow.models import Variable
import json


args = {
     'owner': 'xxx',
     'start_date':'2020-11-5',
     'provide_context': True
    }


dag = DAG(
    dag_id='dynamic',
    default_args=args
    )


def return_bucket_files(bucket_name='xxxxx', **kwargs):
    client = storage.Client()
    bucket = client.get_bucket(bucket_name)
    blobs = bucket.list_blobs()

    file_list = [blob.name for blob in blobs]

    return file_list


def dynamic_gcs_to_gbq_etl(file, **kwargs):

    mapping = json.loads(Variable.get("xxxxx"))
    database = mapping[0][file]
    table = mapping[1][file]

    task=GoogleCloudStorageToBigQueryOperator(
        task_id= f'gcs_load_{file}_to_gbq',
        bucket='xxxxxxx',
        source_objects=[f'{file}'],
        destination_project_dataset_table=f'xxx.{database}.{table}',
        write_disposition="WRITE_TRUNCATE",
        autodetect=True,
        skip_leading_rows=1,
        source_format='CSV',
        dag=dag)

    return task


start_task = DummyOperator(
    task_id='start',
    dag=dag
)


end_task = DummyOperator(
    task_id='end',
    dag=dag)


push_bucket_files = PythonOperator(
        task_id="return_bucket_files",
        provide_context=True,
        python_callable=return_bucket_files,
        dag=dag)


for file in return_bucket_files():
    gcs_load_task = dynamic_gcs_to_gbq_etl(file)
    start_task >> push_bucket_files >> gcs_load_task >> end_task

Answer 1

此问题意味着Web服务器无法在其一侧填充DAG包--这个问题很可能与您的DAG无关。

我的建议是现在尝试重新启动web服务器(通过安装虚拟包)。

类似的问题在本帖子以及这里中也有报道。