Bulk Upsert for MySQL & PostgreSQL

Upsert 소개
"UPSERT"is a DBMS feature that allows a DML statement's author to atomically either insert a row, or on the basis of the row already existing, UPDATE that existing row instead, while safely giving little to no further thought to concurrency. One of those two outcomes must be guaranteed, regardless of concurrent activity, which has been called "the essential property of UPSERT".
요컨대 존재하지 않으면 삽입하고 존재하면 갱신하는 것이다.
단일 레코드 Upsert
MySQL에 INSERT가 있는데...ON DUPLICATE KEY UPDATE 구문을 사용하여 Upsert를 구현할 수 있습니다.

INSERT INTO customers (id, first_name, last_name, email) VALUES (30797, 'hooopo1', 'wang', '[email protected]') 
ON DUPLICATE KEY UPDATE 
first_name = VALUES(first_name), last_name = VALUES(last_name);

PostgreSQL은 9.5에서부터 INSERT까지...ON CONFLICT UPDATE 구문은 MySQL과 유사합니다.

INSERT INTO customers (id, first_name, last_name, email) VALUES (30797, 'hooopo1', 'wang', '[email protected]') 
ON CONFLICT(id) DO  UPDATE 
SET first_name = EXCLUDED.first_name, last_name = EXCLUDED.last_name;

일괄 Upsert
이전에 MySQL에 어떻게 삽입하는 것이 가장 빠른지 연구한 결과 LOAD INFILE 방식으로 대량으로 삽입하는 것을 언급했고 MySQLbulk insert은 지원REPLACE의 의미로 대량으로 삽입하는 동시에 upsert할 수 있다.

LOAD DATA LOCAL INFILE '/Users/hooopo/data/out/product_sales_facts.txt'
REPLACE INTO TABLE product_sale_facts FIELDS TERMINATED BY ',' (`id`,`date_id`,`order_id`,`product_id`,`address_id`,`unit_price`,`purchase_price`,`gross_profit`,`quantity`,`channel_id`,`gift`)

물론 PostgreSQL에도 Copy 기능이 있는데 MySQLLOAD INFILE과 유사하다.그러나 copy 명령은 Upsert를 지원하지 않기 때문에 일부 증량 ETL의 작업이 매우 불편하다.
그러나 Staging 테이블을 이용하여 bulk upsert를 실현하는 방법이 있는데 대체적으로 다음과 같다.
하나.목표 표
둘.증량 데이터를 대량으로 중간표에 삽입하다

CREATE TABLE IF NOT EXISTS staging  LIKE customers INCLUDING DEFAULTS INCLUDING CONSTRAINTS INCLUDING INDEXES;
COPY staging (id, email, first_name, last_name)
        FROM STDIN
          WITH
            DELIMITER ','
            NULL '\N'
            CSV;

셋.목표표에서staging표와 충돌하는 부분을 삭제합니다

DELETE FROM customers
USING staging
WHERE customers.id = staging.id

넷.staging표를 목표표에 대량으로 삽입합니다. 충돌 부분은 삭제되었기 때문에 이 단계는 어떠한 충돌도 없습니다.

INSERT INTO customers (SELECT * FROM staging);

오.Staging 테이블을 비웁니다.

TRUNCATE TABLE staging;

위의 절차는 정말 번거롭습니다. 만약에 키바플러스를 사용한다면 간단한 DSL만 필요합니다.

destination Kiba::Plus::Destination::PgBulk2, { :connect_url => DEST_URL,
                                :table_name => "customers",
                                :truncate => false,
                                :columns => [:id, :email, :first_name, :last_name],
                                :incremental => true,
                                :unique_by => :id
                              }

관련 링크:

https://wiki.postgresql.org/w...

http://www.silota.com/blog/am...

https://hashrocket.com/blog/p...

http://docs.aws.amazon.com/re...

http://stackoverflow.com/ques...