Как исправить ошибку "psycopg2.errors.BadCopyFileFormat: missing data for column"?

При попытке загрузить данные из csv-файла в базу данных Postgres с помощью PostgresHook.bulk_load возникает ошибка:

[2024-08-20, 09:02:01 UTC] {postgres.py:168} INFO - Running copy expert: COPY test_nikita2 FROM STDIN, filename: /opt/airflow/dags/burmistrov/data_nikita/2024-08-19.csv
[2024-08-20, 09:02:01 UTC] {base.py:73} INFO - Using connection ID 'test_db' for task execution.
[2024-08-20, 09:02:01 UTC] {taskinstance.py:1943} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/decorators/base.py", line 221, in execute
    return_value = super().execute(context)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/operators/python.py", line 192, in execute
    return_value = self.execute_callable()
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/operators/python.py", line 209, in execute_callable
    return self.python_callable(*self.op_args, **self.op_kwargs)
  File "/opt/airflow/dags/burmistrov/test.py", line 175, in transfer_data_api
    hook.bulk_load('test_nikita2', fname)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/providers/postgres/hooks/postgres.py", line 192, in bulk_load
    self.copy_expert(f"COPY {table} FROM STDIN", tmp_file)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/providers/postgres/hooks/postgres.py", line 176, in copy_expert
    cur.copy_expert(sql, file)
psycopg2.errors.BadCopyFileFormat: missing data for column "impressions"
CONTEXT:  COPY test_nikita2, line 1: "campaign,impressions,clicks,costs"
[2024-08-20, 09:02:01 UTC] {taskinstance.py:1400} INFO - Marking task as FAILED. dag_id=test_dag, task_id=transfer_data_api, execution_date=20240820T090154, start_date=20240820T090158, end_date=20240820T090201
[2024-08-20, 09:02:01 UTC] {standard_task_runner.py:104} ERROR - Failed to execute job 70948 for task transfer_data_api (missing data for column "impressions"
CONTEXT:  COPY test_nikita2, line 1: "campaign,impressions,clicks,costs"
; 58980)
[2024-08-20, 09:02:01 UTC] {local_task_job_runner.py:228} INFO - Task exited with return code 1

Пример данных в файле:

campaign,impressions,clicks,costs
asd.com_xzc_catvendor,52,27,3042.04
sd-fgh.com_SMEG_k50gen,31,1,15.17

Таблицу создаю так:

sql_schema_init = """
CREATE TABLE IF NOT EXISTS test_nikita2 (
    Campaign text NULL,
    Impressions integer NULL,
    Clicks integer NULL,
    Cost float NULL
);
"""

Данные передаю так:

with open(os.path.join(path+"{}.csv".format(get_dates())), 'w+', encoding='utf-8') as f:
   f.write(resultcsv)                          
   fname = f.name                
try:
   hook = PostgresHook(postgres_conn_id='test_db')
   print(321)
   hook.bulk_load('test_nikita2', fname)                
finally:
   pass

Я пробовал менять разделители, не работает ни с «;», ни с «,». Я также попробовал изменить типы данных в бд на просто текст, чтобы получить там что-нибудь, чтобы понять, в чем проблема, но в таблицу ничего не передается. Также подсказали, что bulk_load не обрабатывает заголовки в csv файлах, я пробовал передавать файл без заголовка, но выдает точно такую же ошибку.


Ответы (1 шт):

Автор решения: Rabban Keyak

Нужно или удалить из csv-файл первую строку, которая

campaign,impressions,clicks,costs

Или дать понять постгресу, что в первой строке заголовок с помощью параметра HEADER инструкции COPY, описание которого есть, например, в этом переводе документации: https://postgrespro.ru/docs/postgresql/16/sql-copy (или его эквивалента в psycopg2).

→ Ссылка