气流中Postgres Hook的CSV_Expert不会在没有错误的情况下写任何东西

发布于 2025-01-25 13:10:30 字数 1545 浏览 3 评论 0原文

我想通过气流将CSV文件写入Postgres的表格。我遇到了这条气流表示挂钩已经具有用于CSV导出的内置函数。并使用此 thread 如何使用它。我有一个Python操作员，其Python_collable如下：

def copy_expert_csv():
    hook = PostgresHook(postgres_conn_id='warehouse',host='data-warehouse',
                        database='datalake',
                        user='root',
                        password='root',
                        port=9999)
    with hook.get_conn() as connection:
            hook.copy_expert("""COPY datalake.public.wcc_users  FROM stdin WITH CSV HEADER
                        DELIMITER as ',' """,
                        'includes/cleaned_data/wwc/' + str(date.today()) + '_wwc_cleaned ')
            connection.commit()

该任务成功完成，如图像所示。：

并且我的数据库中也没有错误日志：

materials-data-warehouse-1  | 2022-04-29 17:43:01.942 UTC [198] STATEMENT:  COPY datalake.public.wcc_users  FROM STDIN WITH (FORMAT CSV) HEADER

我的文件约为1000行。但是，当我从表中选择时，插入了0行。

表中的列命名与文件不同，并且两个列也具有日期和时间戳数据类型，而不是文本。可以是原因吗？那为什么不丢任何错误呢？

原文

I want to write a CSV file to a table in Postgres via Airflow.
I came across this Airflow documentation denoting that the hook already has a builtin function for CSV export.
And used this thread on how to use it.
I have a python operator whose python_callable is as follows:

def copy_expert_csv():
    hook = PostgresHook(postgres_conn_id='warehouse',host='data-warehouse',
                        database='datalake',
                        user='root',
                        password='root',
                        port=9999)
    with hook.get_conn() as connection:
            hook.copy_expert("""COPY datalake.public.wcc_users  FROM stdin WITH CSV HEADER
                        DELIMITER as ',' """,
                        'includes/cleaned_data/wwc/' + str(date.today()) + '_wwc_cleaned ')
            connection.commit()

The task finishes successfully as shown in the image.

:

And there is no error logs on my database either:

materials-data-warehouse-1  | 2022-04-29 17:43:01.942 UTC [198] STATEMENT:  COPY datalake.public.wcc_users  FROM STDIN WITH (FORMAT CSV) HEADER

My file has around 1000 rows. However, when I select from my table, there are 0 rows inserted.

The column naming in table are different from the file and also 2 columns have date and timestamp datatypes rather than text. Can it be the cause? Then why no errors are thrown?

分享到QQ

分享到微博