Airflow官方pet示例DAG插入数据与get_birth_date任务报错排查

1. populate_pet_table任务报错 #

错误原因 #

你创建的pet表第一个字段是pet_id，属于SERIAL类型的整数自增主键，但你执行INSERT语句时没有显式指定插入的列名，PostgreSQL会默认按表定义的字段顺序匹配VALUES中的值，相当于把字符串类型的宠物名直接插入到整数类型的pet_id列，触发类型不匹配错误。

修复方式 #

在所有INSERT语句中显式指定要插入的列名，跳过自增的pet_id字段，让数据库自动生成主键值：

INSERT INTO pet (name, pet_type, birth_date, owner) VALUES ( 'Max', 'Dog', '2018-07-05', 'Jane');
INSERT INTO pet (name, pet_type, birth_date, owner) VALUES ( 'Susie', 'Cat', '2019-05-01', 'Phil');
INSERT INTO pet (name, pet_type, birth_date, owner) VALUES ( 'Lester', 'Hamster', '2020-06-23', 'Lily');
INSERT INTO pet (name, pet_type, birth_date, owner) VALUES ( 'Quincy', 'Parrot', '2013-08-11', 'Anne');

2. get_birth_date任务报错 #

错误原因 #

你在SQL模板中直接使用{{ params.begin_date }}和{{ params.end_date }}，Jinja渲染完成后生成的SQL为BETWEEN SYMMETRIC 2020-01-01 AND 2020-12-31，PostgreSQL会将没有引号包裹的2020-01-01识别为算术运算（2020减1减1），最终得到整数2018，相当于用日期类型的birth_date字段和整数做比较，触发类型匹配错误。

修复方式 #

给两个模板变量加上单引号，让PostgreSQL将其识别为日期字符串：

SELECT * FROM pet
WHERE birth_date
BETWEEN SYMMETRIC '{{ params.begin_date }}' AND '{{ params.end_date }}';

修正后的完整DAG代码 #

import datetime

from airflow import DAG
from airflow.providers.postgres.operators.postgres import PostgresOperator

with DAG(
    dag_id="postgres_operator_dag",
    start_date=datetime.datetime(2020, 2, 2),
    schedule_interval="@once",
    catchup=False,
) as dag:
    create_pet_table = PostgresOperator(
        task_id="create_pet_table",
        postgres_conn_id="postgres_default",
        sql="""
            CREATE TABLE IF NOT EXISTS pet (
            pet_id SERIAL PRIMARY KEY,
            name VARCHAR NOT NULL,
            pet_type VARCHAR NOT NULL,
            birth_date DATE NOT NULL,
            OWNER VARCHAR NOT NULL);
          """,
    )
    populate_pet_table = PostgresOperator(
        task_id="populate_pet_table",
        postgres_conn_id="postgres_default",
        sql="""
            INSERT INTO pet (name, pet_type, birth_date, owner) VALUES ( 'Max', 'Dog', '2018-07-05', 'Jane');
            INSERT INTO pet (name, pet_type, birth_date, owner) VALUES ( 'Susie', 'Cat', '2019-05-01', 'Phil');
            INSERT INTO pet (name, pet_type, birth_date, owner) VALUES ( 'Lester', 'Hamster', '2020-06-23', 'Lily');
            INSERT INTO pet (name, pet_type, birth_date, owner) VALUES ( 'Quincy', 'Parrot', '2013-08-11', 'Anne');
            """,
    )
    get_all_pets = PostgresOperator(
        task_id="get_all_pets", postgres_conn_id="postgres_default", sql="SELECT * FROM pet;"
    )
    get_birth_date = PostgresOperator(
        task_id="get_birth_date",
        postgres_conn_id="postgres_default",
        sql="""
            SELECT * FROM pet
            WHERE birth_date
            BETWEEN SYMMETRIC '{{ params.begin_date }}' AND '{{ params.end_date }}';
            """,
        params={'begin_date': '2020-01-01', 'end_date': '2020-12-31'},
    )

    create_pet_table >> populate_pet_table >> get_all_pets >> get_birth_date

内容的提问来源于stack exchange，提问作者Arie