Ребята, вот имею такую задачу - необходимо скопировать данные из одной схемы БД в другую в этой же БД (БД - PostgreSQL). При этом необходимо соблюдать некоторые условия. А именно:
Insert :
Load table incrementally using project_id as a lookup. Insert project record into this table when project_id does not exist.
Update :
If project_id already exists and the value of other attributes is changed, then update only the changed values for the collect project (project_id).
Пока не до конца понял сами условия, если честно. В связи с чем не могу правильно сформировать запрос. Сейчас запрос как базовый выглядит так:
INSERT INTO target_schema.supplies select * from source_schema.supplies
но уже есть ошибка:
2021-08-10 13:05:23,482] {{taskinstance.py:1128}} ERROR - duplicate key value violates unique constraint "supplies_pkey"
DETAIL: Key (id)=(2) already exists.
Traceback (most recent call last):
File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 966, in _run_raw_task
result = task_copy.execute(context=context)
File "/usr/local/lib/python3.7/site-packages/airflow/operators/python_operator.py", line 113, in execute
return_value = self.execute_callable()
File "/usr/local/lib/python3.7/site-packages/airflow/operators/python_operator.py", line 118, in execute_callable
return self.python_callable(*self.op_args, **self.op_kwargs)
File "/usr/local/airflow/dags/local_copy_source_schema_to_target_schema.py", line 31, in copy_data_from_source_schema_to_target_schema
redshift_hook.run(sql_queries)
File "/usr/local/lib/python3.7/site-packages/airflow/hooks/dbapi_hook.py", line 175, in run
cur.execute(s)
psycopg2.errors.UniqueViolation: duplicate key value violates unique constraint "supplies_pkey"
DETAIL: Key (id)=(2) already exists.
Другие условия даже не включил так как не понял.
Можете подсказать где ошибка в запросе исходя из условий и как доделать запрос чтобы все условия удовлетворялись?
Спасибо.
Пока только начинаю погружаться в SQL и еще есть проблемы.
P.S. Был вопрос - использовать ли ON CONFLICT DO UPDATE или же смотреть только те строки, где были изменения и обновлять, а остальное просто добавлять.
Пока остановился на такой конструкции:
INSERT INTO target_redshift.supplies
SELECT s.*
FROM source_redshift.supplies AS s
WHERE NOT EXISTS (SELECT 1 FROM target_redshift.supplies AS t WHERE s.project_id = t.project_id)
ON conflict
DO nothing;
Вроде так должно быть чуть быстрее, чем делать ON CONFLICT DO UPDATE. Что скажете?
INSERT .. ON CONFLICT
же... Пока не до конца понял сами условия, если честно. А что тут понимать? если нет записи с таким проектом - вставить, если есть - обновить данные в существующей записи. Всё.