气流PythonOperator任务失败类型错误：密钥必须是字符串

Question

这里的新手，请容忍我。。 我不明白为什么这个简单的任务失败了：

def getCarJSON():
    dictCars= {'link': '/cars/acura', 'num': '1'}    
    with open('data/dictCars.json', 'w') as fp:
        json.dump(dictCars, fp)

这是作为JSON存储在磁盘上的简单dict。 为什么我会得到：

Broken DAG: [/home/user/airflow/dags/cars.py] Traceback (most recent call last): File "/usr/local/lib/python3.8/dist-packages/airflow/models/baseoperator.py", line 404, in init validate_key(task_id) File "/usr/local/lib/python3.8/dist-packages/airflow/utils/helpers.py", line 39, in validate_key raise TypeError("The key has to be a string") TypeError: The key has to be a string

我在DAG文件中有常用数据：

# Set default args
default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2021, 3, 23),
    'email': ['donko@gmail.com'],
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 0,
    'retry_delay': timedelta(minutes=2)
}

schedule_interval = '30 09 * * *'

# Define DAG: Set ID and assign default args and schedule interval
dag = DAG(
    dag_id = 'get_cars',
    default_args = default_args,
    schedule_interval = schedule_interval
)


# Get cars dict
get_cars_json = PythonOperator(
    task_id=getCarJSON,
    python_callable=getCarJSON,
    dag=dag
)

我只想在驱动器上转储数据

Answer 1

从断开的dag消息来看，在验证tast_id属性时，似乎出现了此错误

在下面的代码中，您将getCarJSON函数传递给了task_id，而不是传递任务的名称，任务的名称将是字符串类型，因此会导致这个类型错误问题

get_cars_json = PythonOperator(
    task_id='getCarJSON', # Name here was without quotations
    python_callable=getCarJSON,
    dag=dag
)