DBA: 동기화 및 가져오기를 능숙하게 구성합니다.

대규모 데이터 세트의 복잡한 처리용(다양한 ETL 프로세스: 외부 소스와의 가져오기, 변환 및 동기화)가 필요한 경우가 많습니다. 일시적으로 “기억”하고 즉시 신속하게 처리 뭔가 방대한 것.

이런 종류의 일반적인 작업은 일반적으로 다음과 같습니다. "바로 여기에 고객 은행에서 회계 부서가 언로드됨 마지막으로 받은 지불금을 신속하게 웹사이트에 업로드하고 계정에 연결해야 합니다.”

그러나 이 "무언가"의 양이 수백 메가바이트로 측정되기 시작하고 서비스가 연중무휴 24시간 데이터베이스와 계속 작동해야 하면 인생을 망치는 많은 부작용이 발생합니다.

PostgreSQL뿐만 아니라 PostgreSQL에서도 이러한 문제를 처리하려면 모든 것을 더 빠르게 처리하고 리소스를 덜 소비할 수 있는 몇 가지 최적화를 사용할 수 있습니다.

1. 어디로 배송되나요?

먼저, "처리"하려는 데이터를 어디에 업로드할 수 있는지 결정해 보겠습니다.

1.1. 임시 테이블(TEMPORARY TABLE)

원칙적으로 PostgreSQL의 경우 임시 테이블은 다른 테이블과 동일합니다. 그러므로 다음과 같은 미신은 "거기의 모든 것은 메모리에만 저장되며 끝날 수 있습니다.". 그러나 몇 가지 중요한 차이점도 있습니다.

데이터베이스 연결마다 고유한 "네임스페이스"

두 개의 연결이 동시에 연결을 시도하는 경우 CREATE TABLE x, 그러면 누군가는 확실히 얻을 것입니다 비고유성 오류 데이터베이스 객체.

그러나 둘 다 실행을 시도하면 CREATE TEMPORARY TABLE x, 그러면 둘 다 정상적으로 수행하고 모두가 당신의 사본 테이블. 그리고 그들 사이에는 공통점이 없을 것입니다.

연결을 끊을 때 "자기 파괴"

연결이 종료되면 모든 임시 테이블이 자동으로 삭제되므로 수동으로 DROP TABLE x 그거 빼곤 아무 의미가 없어...

당신이 통해 일하고 있다면 트랜잭션 모드의 pgbouncer, 데이터베이스는 이 연결이 여전히 활성 상태라고 계속 믿고 있으며 이 임시 테이블이 여전히 존재합니다.

따라서 다른 연결에서 pgbouncer로 다시 생성하려고 하면 오류가 발생합니다. 하지만 이는 다음을 사용하여 피할 수 있습니다. CREATE TEMPORARY TABLE IF NOT EXISTS x.

사실, 어쨌든 이 작업을 수행하지 않는 것이 좋습니다. 그러면 "이전 소유자"의 남은 데이터를 "갑자기" 찾을 수 있기 때문입니다. 대신, 매뉴얼을 읽고 테이블을 생성할 때 추가가 가능하다는 것을 확인하는 것이 훨씬 좋습니다. ON COMMIT DROP - 즉, 트랜잭션이 완료되면 테이블이 자동으로 삭제됩니다.

비복제

임시 테이블은 특정 연결에만 속하기 때문에 복제되지 않습니다. 하지만 이를 통해 데이터를 이중으로 기록할 필요가 없습니다. 힙 + WAL에 있으므로 INSERT/UPDATE/DELETE가 훨씬 빠릅니다.

그러나 임시 테이블은 여전히 ​​"거의 일반적인" 테이블이므로 복제본에도 생성할 수 없습니다. 적어도 현재로서는 해당 패치가 오랫동안 배포되어 왔습니다.

1.2. 기록되지 않은 테이블

하지만 예를 들어 하나의 트랜잭션 내에서 구현할 수 없는 일종의 번거로운 ETL 프로세스가 있지만 여전히 트랜잭션 모드의 pgbouncer? ..

또는 데이터 흐름이 너무 커서 하나의 연결에 대역폭이 충분하지 않습니다. 데이터베이스에서(읽기, CPU당 하나의 프로세스)?..

아니면 일부 작업이 진행 중입니다. 비동기적으로 다른 연결에서?..

여기에는 단 하나의 옵션이 있습니다. 임시가 아닌 테이블을 임시로 생성. 말장난, 응. 그건:

• 누구와도 교차하지 않도록 최대한 무작위 이름으로 "나만의" 테이블을 만들었습니다.
• 발췌: 외부 소스의 데이터로 채웠습니다.
• 변환: 변환되어 키 연결 필드가 채워짐
• 하중: 준비된 데이터를 대상 테이블에 붓습니다.
• "내" 테이블을 삭제했습니다.

그리고 지금 - 연고에 파리가 있습니다. 사실은, PostgreSQL의 모든 쓰기는 두 번 발생합니다. - WAL에서 처음으로, 그런 다음 테이블/인덱스 본문으로 이동합니다. 이 모든 작업은 ACID를 지원하고 사이의 올바른 데이터 가시성을 지원하기 위해 수행됩니다. COMMIT'고약하고 ROLLBACK'널 거래.

하지만 우리는 이것이 필요하지 않습니다! 우리는 모든 과정을 가지고 있습니다 완전히 성공했거나 그렇지 않았거나 둘 중 하나입니다.. 중간 트랜잭션이 얼마나 많이 발생하는지는 중요하지 않습니다. 특히 그것이 어디에 있었는지 명확하지 않은 경우 "중간에서 프로세스를 계속"하는 데 관심이 없습니다.

이를 위해 PostgreSQL 개발자는 버전 9.1에서 다음과 같은 기능을 도입했습니다. 로그되지 않은 테이블:

이 표시를 사용하면 테이블이 기록되지 않은 상태로 생성됩니다. 기록되지 않은 테이블에 기록된 데이터는 미리 쓰기 로그(29장 참조)를 거치지 않습니다. 평소보다 훨씬 빨리 일해. 그러나 실패로부터 면역되지는 않습니다. 서버 장애 또는 긴급 종료 시 로그되지 않은 테이블 자동으로 잘림. 또한 기록되지 않은 테이블의 내용 복제되지 않음 슬레이브 서버에. 기록되지 않은 테이블에 생성된 모든 인덱스는 자동으로 기록되지 않습니다.

요컨대, 훨씬 더 빠를 거예요, 그러나 데이터베이스 서버가 "떨어지면" 불쾌할 것입니다. 그러나 이런 일이 얼마나 자주 발생하며, ETL 프로세스는 데이터베이스를 "활성화"한 후 "중간에서" 이를 올바르게 수정하는 방법을 알고 있습니까?..

그렇지 않은 경우 위의 경우가 귀하의 경우와 유사하면 다음을 사용하십시오. UNLOGGED하지만 결코 실제 테이블에서는 이 속성을 활성화하지 마십시오., 귀하에게 소중한 데이터입니다.

1.3. 커밋 시 { 행 삭제 | 떨어지다}

이 구성을 사용하면 테이블 생성 시 트랜잭션이 완료될 때 자동 동작을 지정할 수 있습니다.

약 ON COMMIT DROP 위에서 이미 썼는데, 생성됩니다. DROP TABLE, 하지만 함께 ON COMMIT DELETE ROWS 상황이 더 흥미롭습니다. 여기서 생성됩니다. TRUNCATE TABLE.

임시 테이블의 메타 설명을 저장하기 위한 전체 인프라는 일반 테이블과 완전히 동일하므로 임시 테이블의 지속적인 생성 및 삭제로 인해 시스템 테이블이 심각하게 "부풀어 오름" pg_class, pg_attribute, pg_attrdef, pg_dependent,…

이제 매초마다 새 트랜잭션을 열고 임시 테이블을 생성, 채우기, 처리 및 삭제하는 데이터베이스에 직접 연결된 작업자가 있다고 상상해 보십시오. 시스템 테이블에 과도한 쓰레기가 축적될 것이며, 이로 인해 각 작업마다 추가 브레이크가 발생합니다.

일반적으로 이렇게 하지 마세요! 이런 경우에는 훨씬 더 효과적입니다. CREATE TEMPORARY TABLE x ... ON COMMIT DELETE ROWS 트랜잭션 주기에서 벗어나십시오. 그러면 각각의 새로운 트랜잭션이 시작될 때마다 테이블이 이미 존재할 것이다 (통화 저장 CREATE),하지만 비어있을 것이다, 덕분에 TRUNCATE (저희는 해당 호출도 저장했습니다) 이전 거래를 완료할 때.

1.4. 다음을 포함하여...

나는 처음에 임시 테이블의 일반적인 사용 사례 중 하나가 다양한 종류의 가져오기라고 언급했습니다. 개발자는 지쳐서 대상 테이블의 필드 목록을 임시 테이블 선언에 복사하여 붙여넣습니다.

그러나 게으름은 발전의 원동력입니다! 그렇기 때문에 "샘플을 기반으로" 새 테이블 만들기 훨씬 더 간단할 수 있습니다.

CREATE TEMPORARY TABLE import_table(
  LIKE target_table
);

그러면 이 테이블에 많은 데이터를 생성할 수 있으므로 검색 속도가 결코 빨라지지 않습니다. 그러나 이에 대한 전통적인 해결책이 있습니다. 바로 인덱스입니다! 그리고 그렇습니다. 임시 테이블에도 인덱스가 있을 수 있습니다.

필요한 인덱스가 대상 테이블의 인덱스와 일치하는 경우가 많기 때문에 간단히 작성할 수 있습니다. LIKE target_table INCLUDING INDEXES.

당신도 필요하다면 DEFAULT-값(예: 기본 키 값을 입력하기 위해)을 사용할 수 있습니다. LIKE target_table INCLUDING DEFAULTS. 아니면 간단히 - LIKE target_table INCLUDING ALL — 기본값, 인덱스, 제약 조건 등을 복사합니다.

하지만 여기서 당신이 만든 경우 인덱스를 사용하여 즉시 테이블을 가져오면 데이터를 로드하는 데 시간이 더 오래 걸립니다.먼저 모든 것을 채운 다음 인덱스를 롤업하는 것보다 - 예를 들어 이것이 어떻게 수행되는지 살펴보십시오. pg_dump.

일반적으로, RTFM!

2. 어떻게 작성하나요?

그냥 말해두죠 - 사용해 보세요 COPY- "팩" 대신 흐름 INSERT, 때때로 가속. 미리 생성된 파일에서 직접 사용할 수도 있습니다.

3. 처리 방법은 무엇입니까?

이제 소개를 다음과 같이 해보자.

• 데이터베이스에 클라이언트 데이터가 저장된 테이블이 있습니다. 1만 개의 레코드
• 매일 클라이언트가 당신에게 새로운 것을 보냅니다 전체 "이미지"
• 경험을 통해 당신은 때때로 그것을 알고 있습니다 10개 이하의 레코드가 변경됩니다.

그러한 상황의 전형적인 예는 다음과 같습니다. KLADR 베이스 — 전체적으로 주소는 많지만 매주 업로드할 때마다 전국 규모에서도 변경 사항(정착지 이름 변경, 거리 결합, 새 집 출현)이 거의 없습니다.

3.1. 전체 동기화 알고리즘

단순화를 위해 데이터를 재구성할 필요조차 없다고 가정해 보겠습니다. 테이블을 원하는 형식으로 가져오면 됩니다.

• 제거하다 더 이상 존재하지 않는 모든 것
• 새로 고침 이미 존재하고 업데이트가 필요한 모든 것
• 삽입 아직 일어나지 않은 모든 일

왜 이 순서로 작업을 수행해야 합니까? 이렇게 하면 테이블 크기가 최소한으로 커지기 때문입니다(MVCC를 기억하세요!).

dst에서 삭제

아니요. 물론 다음 두 가지 작업만 수행하면 됩니다.

• 제거하다 (DELETE) 일반적으로 모든 것
• 삽입 모두 새로운 이미지에서

하지만 동시에 MVCC 덕분에 테이블 크기가 정확히 두 배로 늘어납니다.! 1K 업데이트로 인해 테이블에 있는 레코드의 +10M 이미지를 얻는 것은 너무나 중복됩니다...

TRUNCATE dst

경험이 많은 개발자는 전체 태블릿을 매우 저렴하게 청소할 수 있다는 것을 알고 있습니다.

• 분명히하다 (TRUNCATE) 전체 테이블
• 삽입 모두 새로운 이미지에서

방법은 효과적이며, 때로는 꽤 적용 가능, 하지만 문제가 있습니다... 오랫동안 1만 개의 레코드를 추가할 예정이므로 이 시간 동안 테이블을 비워 둘 여유가 없습니다(단일 트랜잭션으로 래핑하지 않으면 발생함).

이는 다음을 의미합니다.

• 우리는 시작해요 장기 실행 트랜잭션
• TRUNCATE 부과하다 액세스 독점-블로킹
• 우리는 오랫동안 삽입을 하고, 다른 사람들도 이 시간에 심지어 할 수 없다 SELECT

뭔가 잘 안되네요...

테이블 변경… 이름 바꾸기… / 테이블 삭제…

대안은 모든 것을 별도의 새 테이블에 채운 다음 이전 테이블 대신 이름을 바꾸는 것입니다. 몇 가지 불쾌한 작은 것들:

• 아직도 액세스 독점, 비록 시간은 훨씬 짧았지만
• 이 테이블에 대한 모든 쿼리 계획/통계가 재설정됩니다. ANALYZE를 실행해야 합니다
• 모든 외래 키가 손상되었습니다. (FK) 테이블로

Simon Riggs가 제안한 WIP 패치가 있었습니다. ALTER-통계 및 FK를 건드리지 않고 파일 수준에서 테이블 본문을 바꾸는 작업이지만 쿼럼을 수집하지는 않았습니다.

삭제, 업데이트, 삽입

그래서 우리는 세 가지 작업의 비차단 옵션을 선택했습니다. 거의 XNUMX개... 이 작업을 가장 효과적으로 수행하는 방법은 무엇입니까?

-- все делаем в рамках транзакции, чтобы никто не видел "промежуточных" состояний
BEGIN;

-- создаем временную таблицу с импортируемыми данными
CREATE TEMPORARY TABLE tmp(
  LIKE dst INCLUDING INDEXES -- по образу и подобию, вместе с индексами
) ON COMMIT DROP; -- за рамками транзакции она нам не нужна

-- быстро-быстро вливаем новый образ через COPY
COPY tmp FROM STDIN;
-- ...
-- .

-- удаляем отсутствующие
DELETE FROM
  dst D
USING
  dst X
LEFT JOIN
  tmp Y
    USING(pk1, pk2) -- поля первичного ключа
WHERE
  (D.pk1, D.pk2) = (X.pk1, X.pk2) AND
  Y IS NOT DISTINCT FROM NULL; -- "антиджойн"

-- обновляем оставшиеся
UPDATE
  dst D
SET
  (f1, f2, f3) = (T.f1, T.f2, T.f3)
FROM
  tmp T
WHERE
  (D.pk1, D.pk2) = (T.pk1, T.pk2) AND
  (D.f1, D.f2, D.f3) IS DISTINCT FROM (T.f1, T.f2, T.f3); -- незачем обновлять совпадающие

-- вставляем отсутствующие
INSERT INTO
  dst
SELECT
  T.*
FROM
  tmp T
LEFT JOIN
  dst D
    USING(pk1, pk2)
WHERE
  D IS NOT DISTINCT FROM NULL;

COMMIT;

3.2. 후처리 가져오기

동일한 KLADR에서 변경된 모든 레코드는 정규화, 키워드 강조 표시 및 필요한 구조로 축소 등의 사후 처리를 통해 추가로 실행되어야 합니다. 그런데 어떻게 알죠- 정확히 무엇이 바뀌었나요?동기화 코드를 복잡하게 하지 않고, 이상적으로는 전혀 건드리지 않고요?

동기화 시 프로세스에만 쓰기 액세스 권한이 있는 경우 모든 변경 사항을 수집하는 트리거를 사용할 수 있습니다.

-- целевые таблицы
CREATE TABLE kladr(...);
CREATE TABLE kladr_house(...);

-- таблицы с историей изменений
CREATE TABLE kladr$log(
  ro kladr, -- тут лежат целые образы записей старой/новой
  rn kladr
);

CREATE TABLE kladr_house$log(
  ro kladr_house,
  rn kladr_house
);

-- общая функция логирования изменений
CREATE OR REPLACE FUNCTION diff$log() RETURNS trigger AS $$
DECLARE
  dst varchar = TG_TABLE_NAME || '$log';
  stmt text = '';
BEGIN
  -- проверяем необходимость логгирования при обновлении записи
  IF TG_OP = 'UPDATE' THEN
    IF NEW IS NOT DISTINCT FROM OLD THEN
      RETURN NEW;
    END IF;
  END IF;
  -- создаем запись лога
  stmt = 'INSERT INTO ' || dst::text || '(ro,rn)VALUES(';
  CASE TG_OP
    WHEN 'INSERT' THEN
      EXECUTE stmt || 'NULL,$1)' USING NEW;
    WHEN 'UPDATE' THEN
      EXECUTE stmt || '$1,$2)' USING OLD, NEW;
    WHEN 'DELETE' THEN
      EXECUTE stmt || '$1,NULL)' USING OLD;
  END CASE;
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

이제 동기화를 시작하기 전에 트리거를 적용하거나 다음을 통해 활성화할 수 있습니다. ALTER TABLE ... ENABLE TRIGGER ...):

CREATE TRIGGER log
  AFTER INSERT OR UPDATE OR DELETE
  ON kladr
    FOR EACH ROW
      EXECUTE PROCEDURE diff$log();

CREATE TRIGGER log
  AFTER INSERT OR UPDATE OR DELETE
  ON kladr_house
    FOR EACH ROW
      EXECUTE PROCEDURE diff$log();

그런 다음 로그 테이블에서 필요한 모든 변경 사항을 차분하게 추출하고 추가 핸들러를 통해 실행합니다.

3.3. 연결된 세트 가져오기

위에서는 소스와 대상의 데이터 구조가 동일한 경우를 고려했습니다. 하지만 외부 시스템에서 업로드한 파일의 형식이 우리 데이터베이스의 저장 구조와 다른 경우에는 어떻게 될까요?

고전적인 "다대일" 옵션인 클라이언트 및 해당 계정의 스토리지를 예로 들어 보겠습니다.

CREATE TABLE client(
  client_id
    serial
      PRIMARY KEY
, inn
    varchar
      UNIQUE
, name
    varchar
);

CREATE TABLE invoice(
  invoice_id
    serial
      PRIMARY KEY
, client_id
    integer
      REFERENCES client(client_id)
, number
    varchar
, dt
    date
, sum
    numeric(32,2)
);

그러나 외부 소스로부터의 다운로드는 "올인원" 형태로 제공됩니다.

CREATE TEMPORARY TABLE invoice_import(
  client_inn
    varchar
, client_name
    varchar
, invoice_number
    varchar
, invoice_dt
    date
, invoice_sum
    numeric(32,2)
);

분명히 이 버전에서는 고객 데이터가 복제될 수 있으며 주요 기록은 "계정"입니다.

0123456789;Вася;A-01;2020-03-16;1000.00
9876543210;Петя;A-02;2020-03-16;666.00
0123456789;Вася;B-03;2020-03-16;9999.00

모델의 경우 간단히 테스트 데이터를 삽입하지만 다음을 기억하세요. COPY 더 효율적입니다!

INSERT INTO invoice_import
VALUES
  ('0123456789', 'Вася', 'A-01', '2020-03-16', 1000.00)
, ('9876543210', 'Петя', 'A-02', '2020-03-16', 666.00)
, ('0123456789', 'Вася', 'B-03', '2020-03-16', 9999.00);

먼저, "사실"이 참조하는 "컷"을 강조하겠습니다. 우리의 경우 송장은 고객을 나타냅니다.

CREATE TEMPORARY TABLE client_import AS
SELECT DISTINCT ON(client_inn)
-- можно просто SELECT DISTINCT, если данные заведомо непротиворечивы
  client_inn inn
, client_name "name"
FROM
  invoice_import;

계정을 고객 ID와 올바르게 연결하려면 먼저 이러한 식별자를 찾거나 생성해야 합니다. 그 아래에 필드를 추가해 보겠습니다.

ALTER TABLE invoice_import ADD COLUMN client_id integer;
ALTER TABLE client_import ADD COLUMN client_id integer;

위에서 설명한 테이블 동기화 방법을 약간 수정하여 사용해 보겠습니다. 클라이언트를 "추가 전용"으로 가져오기 때문에 대상 테이블의 어떤 것도 업데이트하거나 삭제하지 않습니다.

-- проставляем в таблице импорта ID уже существующих записей
UPDATE
  client_import T
SET
  client_id = D.client_id
FROM
  client D
WHERE
  T.inn = D.inn; -- unique key

-- вставляем отсутствовавшие записи и проставляем их ID
WITH ins AS (
  INSERT INTO client(
    inn
  , name
  )
  SELECT
    inn
  , name
  FROM
    client_import
  WHERE
    client_id IS NULL -- если ID не проставился
  RETURNING *
)
UPDATE
  client_import T
SET
  client_id = D.client_id
FROM
  ins D
WHERE
  T.inn = D.inn; -- unique key

-- проставляем ID клиентов у записей счетов
UPDATE
  invoice_import T
SET
  client_id = D.client_id
FROM
  client_import D
WHERE
  T.client_inn = D.inn; -- прикладной ключ

사실 다 들어있어요 invoice_import 이제 연락처 필드가 채워졌습니다. client_id, 송장을 삽입합니다.

출처 : habr.com