DBA: 同期とインポートを適切に組織する

大規模なデータセットの複雑な処理（さまざまな ETLプロセス: インポート、変換、外部ソースとの同期) が必要になることがよくあります。 一時的に「記憶」し、すぐに迅速に処理する 何かボリュームのあるもの。

この種の典型的なタスクは通常、次のようになります。 「ここだよ 会計部門が顧客の銀行から降ろされる 最後に受け取った支払いをすぐにウェブサイトにアップロードし、アカウントにリンクする必要があります。」

しかし、この「何か」の量が数百メガバイトに達し始め、サービスが 24 時間年中無休でデータベースと連携し続けなければならない場合、生活を台無しにする多くの副作用が発生します。

PostgreSQL (だけではありません) でそれらに対処するには、いくつかの最適化を使用して、すべてをより高速に、より少ないリソース消費で処理できるようにします。

1. どこに発送しますか?

まず、「処理」したいデータをどこにアップロードできるかを決めましょう。

1.1. 一時テーブル (TEMPORARY TABLE)

原則として、PostgreSQL の一時テーブルは他のものと同じです。 したがって、次のような迷信は、 「そこにあるものはすべて記憶の中にのみ保存され、それは終わる可能性がある」。 しかし、いくつかの重要な違いもあります。

データベースへの接続ごとに独自の「名前空間」

XNUMX つの接続が同時に接続しようとした場合 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. ログに記録されていないテーブル

しかし、たとえば、XNUMX つのトランザクション内で実装できないある種の面倒な ETL プロセスがあるにもかかわらず、まだ トランザクションモードのpgbouncer？..

または、データ フローが大きすぎるため、 XNUMX つの接続で十分な帯域幅がありません データベースから (読み取り、CPU ごとに XNUMX つのプロセス)?

または、何らかの操作が進行中です 非同期的に 別の接続で?.

ここでの選択肢は XNUMX つだけです - 一時的に非一時テーブルを作成する。 しゃれ、そうだね。 あれは：

• 誰とも交差しないように、最大​​限ランダムな名前を付けて「独自の」テーブルを作成しました
• エキス: 外部ソースからのデータを埋め込みました
• 最適化の適用: 変換され、キーリンクフィールドに入力されました
• 負荷: 準備ができたデータをターゲットテーブルに注ぎ込みました
• 「私の」テーブルを削除しました

そして今 - 軟膏の中のハエ。 実際には、 PostgreSQL での書き込みはすべて XNUMX 回行われます - WALで初めて、次にテーブル/インデックス本体に挿入します。 これはすべて、ACID をサポートし、ACID 間のデータの可視性を修正するために行われます。 COMMIT「ナッツと ROLLBACK'ヌルトランザクション。

でも、そんなものは必要ありません！ プロセス全体は私たちにあります それは完全に成功したか、そうでなかったかのどちらかです。。 中間トランザクションがいくつあるかは問題ではありません。特に、どこで行われたかが明確でない場合、「途中からプロセスを続行する」ことには興味がありません。

これを行うために、PostgreSQL 開発者はバージョン 9.1 で次のようなものを導入しました。 ログに記録されていないテーブル:

この指示により、テーブルはログが記録されないものとして作成されます。 ログが記録されていないテーブルに書き込まれたデータは、先行書き込みログ (第 29 章を参照) を通過しないため、そのようなテーブルは いつもよりずっと早く仕事をする。 ただし、失敗を免れないわけではありません。 サーバー障害または緊急シャットダウンの場合、ログに記録されていないテーブル 自動的に切り捨てられる。 さらに、ログに記録されていないテーブルの内容 複製されていない スレーブサーバーに。 ログが記録されないテーブルに作成されたインデックスは、自動的にログが記録されなくなります。

要するに、 はるかに速くなります, しかし、データベースサーバーが「落ちた」場合は不快です。 しかし、これはどのくらいの頻度で起こるのでしょうか?また、ETL プロセスは、データベースを「再活性化」した後、これを「途中から」正しく修正する方法を知っていますか?

そうでない場合、および上記のケースがあなたの場合と似ている場合は、次を使用します。 UNLOGGEDしかし決して 実際のテーブルではこの属性を有効にしないでください、あなたにとって大切なデータ。

1.3. ON COMMIT { 行を削除 | 落とす}

この構造を使用すると、テーブルの作成時にトランザクションが完了したときの自動動作を指定できます。

オン ON COMMIT DROP すでに上に書きましたが、生成されます DROP TABLE、しかし、 ON COMMIT DELETE ROWS 状況はさらに興味深いです - それはここで生成されます TRUNCATE TABLE.

一時テーブルのメタ記述を格納するインフラストラクチャ全体は通常のテーブルのインフラストラクチャとまったく同じであるため、 一時テーブルの継続的な作成と削除は、システム テーブルの深刻な「膨張」につながります pg_class、pg_attribute、pg_attrdef、pg_depend、…

ここで、データベースに直接接続しているワーカーがあり、毎秒新しいトランザクションを開き、一時テーブルを作成、書き込み、処理、削除していると想像してください。システム テーブルには過剰なガベージが蓄積されます。これにより、操作ごとに余分なブレーキが発生します。

一般に、これは行わないでください。 この場合はより効果的です CREATE TEMPORARY TABLE x ... ON COMMIT DELETE ROWS それをトランザクション サイクルから外します。そうすると、新しいトランザクションが開始されるたびに、テーブルはすでに 存在するだろう (通話を保存 CREATE）、 だが 空になります、благодаря TRUNCATE (その呼び出しも保存しました) 前のトランザクションを完了したとき。

1.4. たとえば...を含む...

冒頭で、一時テーブルの典型的な使用例の XNUMX つは、さまざまな種類のインポートであると述べました。開発者は、ターゲット テーブルのフィールドのリストを、一時テーブルの宣言にコピーアンドペーストするのに苦労します。

しかし、怠惰は進歩の原動力です。 それが理由です 「サンプルに基づいて」新しいテーブルを作成します もっと簡単にすることもできます:

CREATE TEMPORARY TABLE import_table(
  LIKE target_table
);

このテーブルに大量のデータを生成できるため、テーブル全体の検索は決して高速ではありません。 しかし、これには伝統的な解決策があります - インデックスです! はい、 一時テーブルにもインデックスを含めることができます.

多くの場合、必要なインデックスはターゲットテーブルのインデックスと一致するため、単純に次のように書くことができます。 LIKE target_table INCLUDING INDEXES.

こちらも必要な場合は DEFAULT-values (たとえば、主キーの値を入力する場合)、次を使用できます。 LIKE target_table INCLUDING DEFAULTS。 あるいは単に - LIKE target_table INCLUDING ALL — デフォルト、インデックス、制約などをコピーします。

ただし、ここで理解する必要があるのは、作成した場合 インデックスを使用してテーブルをすぐにインポートすると、データのロードに時間がかかります最初にすべてを埋めてからインデックスをロールアップする場合よりも、例としてこれがどのように行われるかを見てください。 pg_dump.

全体的に、 RTFM!

2.どうやって書くの？

ただ言っておきます - 使ってください COPY-「パック」の代わりにフロー INSERT, 時々加速。 事前に生成されたファイルから直接行うこともできます。

3. どのように処理するのですか?

それでは、イントロを次のようにしてみましょう。

• データベースにクライアントデータを格納したテーブルがある場合 1万件のレコード
• 毎日、クライアントから新しいものが送られてきます 完全な「イメージ」
• 経験から、あなたは時々それを知っています 変更されるレコードは 10 件までです

このような状況の典型的な例は次のとおりです。 KLADR基地 — 合計すると多数の住所がありますが、毎週のアップロードでは全国規模でもほとんど変更 (集落の名前の変更、通りの結合、新しい家の外観) はほとんどありません。

3.1. 完全同期アルゴリズム

話を簡単にするために、データを再構築する必要さえないとします。テーブルを目的の形式にするだけです。つまり、次のようになります。

• 除去する もう存在しないすべてのもの
• リフレッシュする すでに存在し、更新する必要があるものすべて
• вставить まだ起こっていないことすべて

なぜこの順序で操作を実行する必要があるのでしょうか? これにより、テーブル サイズの増加が最小限に抑えられるため (MVCCを思い出してください！).

dst から削除

いいえ、もちろん、次の XNUMX つの操作だけで済みます。

• 除去する (DELETE）一般的にすべて
• вставить すべては新しいイメージから

しかし同時に、MVCC のおかげで、 テーブルのサイズはちょうど XNUMX 倍になります！ 1K の更新によりテーブル内のレコードの +10 万イメージを取得するのは、まあまあの冗長性です...

dst の切り捨て

経験豊富な開発者は、タブレット全体を非常に安価にクリーニングできることを知っています。

• クリア (TRUNCATE) テーブル全体
• вставить すべては新しいイメージから

という方法が効果的で、 時にはかなり当てはまる, しかし、問題があります... 1 万件のレコードを長期間追加することになるため、テーブルをずっと空のままにしておくわけにはいきません (単一のトランザクションでラップしないとそうなります)。

つまり:

• 始めます 長時間実行されるトランザクション
• TRUNCATE 課す アクセス専用-ブロッキング
• 私たちは長い間挿入を行いますが、他の人はこの時点で それすらできない SELECT

何かがうまくいきません...

ALTER TABLE… RENAME… / DROP TABLE…

別の方法としては、すべてを別の新しいテーブルに入力し、古いテーブルの名前を変更するだけです。 厄介な些細なことがいくつかあります:

• まだすぎる アクセス専用時間は大幅に短縮されますが、
• このテーブルのすべてのクエリ プラン/統計がリセットされます。 ANALYZEを実行する必要があります
• すべての外部キーが壊れています (FK) テーブルへ

Simon Riggs からの WIP パッチがあり、 ALTER-統計やFKには触れずにファイルレベルでテーブル本体を置き換える操作ですが、クォーラムは収集されませんでした。

削除、更新、挿入

したがって、XNUMX つの操作のノンブロッキング オプションに落ち着きます。 もうすぐ 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. リンクされたセットのインポート

上記では、送信元と宛先のデータ構造が同じ場合を検討しました。 しかし、外部システムからのアップロードの形式がデータベースのストレージ構造と異なる場合はどうなるでしょうか?

例として、クライアントとそのアカウントのストレージ、古典的な「多対 XNUMX」オプションを考えてみましょう。

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