气流Pythonoperator任务失败-TypeError：键必须是字符串

Question

气流新手在这里，忍受我...我不明白为什么这个简单的任务失败了：

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

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

断开的DAG:[/home/user/aiffair/dags/cars.py]回溯（最后一次调用）：文件“/usr/local/lib/python3.8/dist packages/aiffair/models/baseoperator.py”，init validate\u key（任务id）文件“/usr/local/lib/python3.8/dist packages/aiffairs/utils/helpers.py”第39行，在validate\key raise raise TypeError（“密钥必须是字符串”）中的第404行TypeError:键必须是字符串

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

# Set default args
default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2021, 3, 23),
    'email': ['[e-mail]'],
    '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\u id，而不是任务的名称，任务的名称将是字符串类型，因此会导致类型错误问题

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