DataWorks:PostgreSQL

オフライン読み取りおよび書き込み

• PostgreSQL データソースは、SCRAM-SHA-256 方式を含むパスワード認証をサポートしています。PostgreSQL データベースでパスワードまたは認証方式を変更した場合、データソースの構成を更新し、接続テストを再度実行したうえで、関連するタスクを手動で実行して変更内容を検証する必要があります。

• PostgreSQL のテーブル名またはフィールド名が数字で始まる場合、大文字と小文字を区別する場合、またはハイフン (-) を含む場合、その名前を二重引用符 ("") で囲む必要があります。それ以外の場合、PostgreSQL プラグインはデータを読み取ったり書き込んだりできません。PostgreSQL Reader および Writer プラグインでは、二重引用符 ("") は JSON のキーワードです。そのため、二重引用符をバックスラッシュ (\) でエスケープする必要があります。たとえば、テーブル名が 123Test の場合、エスケープされた名前は \"123Test\" となります。

  説明

  • 開始および終了の二重引用符の両方をバックスラッシュ (\) でエスケープする必要があります。

  • コードレス UI ではエスケープはサポートされていません。エスケープを行うには、コードエディタに切り替える必要があります。

  以下のコードは、コードエディタでのエスケープ文字の使用例です。

  "parameter": {
      "datasource": "abc",
      "column": [
          "id",
          "\"123Test\"", // エスケープ文字を追加します。
  ],
  "where": "",
  "splitPk": "id",
  "table": "public.wpw_test"
  },

• 一意なインデックスに基づく PostgreSQL データソースのデータ更新はサポートされていません。データを更新するには、まずデータを一時テーブルに書き込んでから、RENAME 操作を使用します。

リアルタイム読み取り

Data Integration におけるリアルタイム同期タスクには、以下の制限が適用されます。

• Data Integration は、ADD COLUMN に対して特別なサポートを提供します。

  • 制約：単一トランザクション内で、ADD COLUMN 操作を他の ADD COLUMN、DROP COLUMN、またはその他のデータ定義言語 (DDL) ステートメントと組み合わせることはできません。

    重要

    ADD COLUMN を DROP COLUMN または RENAME COLUMN などの ALTER COLUMN 動作と併用すると、データ同期タスクが失敗します。

  • 制限：Data Integration は、ADD COLUMN 以外の DDL 操作を検出できません。

• ALTER TABLE および CREATE TABLE 操作はサポートされていません。

• TEMPORARY、UNLOGGED、および Hyper テーブルのレプリケーションはサポートされていません。PostgreSQL には、これらのタイプのテーブルのログを解析してサブスクライブする仕組みが提供されていません。

• Sequences (serial/bigserial/identity) のレプリケーションはサポートされていません。

• TRUNCATE 操作はサポートされていません。

• ラージオブジェクト (Bytea) のレプリケーションはサポートされていません。

• ビュー、マテリアライズドビュー、および外部テーブルのレプリケーションはサポートされていません。

• PostgreSQL を単一テーブルまたは全データベースのリアルタイム同期のソースとして使用する場合、構成済みのデータベースユーザーが所有するテーブルのみを同期できます。

準備 1：アカウントの作成および権限の構成

後続の操作用にデータベースログインアカウントを作成します。このアカウントには、REPLICATION 権限および LOGIN 権限が必要です。

1. アカウントを作成します。

   詳細については、「アカウントおよびデータベースの作成」をご参照ください。

2. 権限を構成します。

   アカウントに replication 権限があるかどうかを確認します。

   select userepl from pg_user where usename='xxx'

   コマンドの戻り値が True の場合、権限が付与されています。False の場合は、権限が付与されていません。権限を付与するには、以下のステートメントを実行します。

   ALTER USER <user> REPLICATION;

準備 2：セカンダリデータベースのサポート状況の確認

SELECT pg_is_in_recovery()

プライマリデータベースのみがサポートされています。コマンドの戻り値は False である必要があります。True の場合は、そのデータベースがセカンダリデータベースであることを意味します。リアルタイム同期はセカンダリデータベースをサポートしていません。データソースの構成を変更して、プライマリデータベースを使用する必要があります。詳細については、「PostgreSQL データソースの構成」をご参照ください。

準備 3：wal_level が logical に設定されているかの確認

show wal_level

wal_level パラメーターは、wal_log のレベルを指定します。コマンドの戻り値は logical である必要があります。それ以外の場合は、論理レプリケーション機構がサポートされていません。

準備 4：wal_sender プロセスの起動可否の確認

-- max_wal_senders を照会します。
show max_wal_senders;

-- pg_stat_replication プロセスの数を照会します。
select count(*) from pg_stat_replication

max_wal_senders の値が空でなく、max_wal_senders の値が pg_stat_replication のエントリ数より大きい場合、アイドル状態の wal_sender プロセスが利用可能です。PostgreSQL データベースは、データ同期プログラムがログをサブスクライバーに送信するために、wal_sender プロセスを起動します。

同期対象の各テーブルについて、ALTER TABLE [tableName] REPLICA IDENTITY FULL ステートメントを実行して、必要な権限を付与する必要があります。それ以外の場合、リアルタイム同期タスクでエラーが報告されます。

リアルタイム PostgreSQL 同期タスクが開始されると、データベース内に自動的にスロットおよびパブリケーションが作成されます。スロット名は di_slot_ + ソリューション ID の形式であり、パブリケーション名は di_pub_ + ソリューション ID の形式です。リアルタイム同期タスクが停止または非公開になった後は、スロットおよびパブリケーションを手動で削除する必要があります。そうしないと、PostgreSQL の Write-Ahead Logging (WAL) が継続して増加する可能性があります。

単一テーブルのオフライン同期タスクの設定ガイド

• 詳細については、「コードレス UI の使用」および「コードエディタの使用」をご参照ください。

• コードエディタ向けのパラメーター一覧およびスクリプト例の完全なリストについては、「付録 1：スクリプト例およびパラメーターの説明」をご参照ください。

全データベースのオフライン読み取りおよびリアルタイム同期タスクの設定ガイド

詳細については、「全データベースのリアルタイム同期タスクの設定」をご参照ください。

コードエディタを使用したバッチ同期タスクの構成

コードエディタを使用してバッチ同期タスクを構成する場合、統一されたスクリプトフォーマット要件に従って、スクリプト内の関連パラメーターを構成する必要があります。詳細については、「コードエディタの使用」をご参照ください。以下では、コードエディタを使用してバッチ同期タスクを構成する際に、データソースに対して構成する必要があるパラメーターについて説明します。

Reader スクリプト例

PostgreSQL データベースからデータを同期および抽出するジョブを構成するには、「コードエディタの使用」の手順に従います。

{
    "type":"job",
    "version":"2.0",// バージョン番号。
    "steps":[
        {
            "stepType":"postgresql",// プラグイン名。
            "parameter":{
                "datasource":"",// データソース。
                "column":[// フィールド。
                    "col1",
                    "col2"
                ],
                "where":"",// フィルター条件。
                "splitPk":"",// データシャーディングに使用するフィールド。データ同期は並行タスクを起動してデータを同期します。
                "table":""// テーブル名。
            },
            "name":"Reader",
            "category":"reader"
        },
        { 
            "stepType":"stream",
            "parameter":{},
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting":{
        "errorLimit":{
            "record":"0"// エラーとなるレコード数。
        },
        "speed":{
            "throttle":true, // throttle が false の場合、mbps パラメーターは無効となり、速度制限は無効になります。throttle が true の場合、速度制限が有効になります。
            "concurrent":1, // 並行ジョブ数。
            "mbps":"12"// 速度制限率。1 mbps は 1 MB/s に相当します。
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

Reader スクリプトパラメーター

Writer スクリプト例

以下のコードは、スクリプト構成の例です。詳細については、以下のパラメーターの説明をご参照ください。

{
  "type":"job",
  "version":"2.0",// バージョン番号。
  "steps":[ 
    {
      "stepType":"stream",
      "parameter":{},
      "name":"Reader",
      "category":"reader"
    },
    {
      "stepType":"postgresql",// プラグイン名。
      "parameter":{
        "datasource":"",// データソース。
        "column":[// フィールド。
          "col1",
          "col2"
        ],
        "table":"",// テーブル名。
        "preSql":[],// データ同期タスク実行前に実行する SQL ステートメント。
        "postSql":[],// データ同期タスク実行後に実行する SQL ステートメント。
      },
      "name":"Writer",
      "category":"writer"
    }
  ],
  "setting":{
    "errorLimit":{
      "record":"0"// エラーとなるレコード数。
    },
    "speed":{
      "throttle":true,// throttle が false の場合、mbps パラメーターは無効となり、速度制限は無効になります。throttle が true の場合、速度制限が有効になります。
      "concurrent":1, // 並行ジョブ数。
      "mbps":"12"// 速度制限率。1 mbps は 1 MB/s に相当します。
    }
  },
  "order":{
    "hops":[
      {
        "from":"Reader",
        "to":"Writer"
      }
    ]
  }
}

Writer スクリプトパラメーター

{
    "job":
    {
        "content":
        [{
            "reader": {...},
            "writer":
            {
                "parameter":
                {
                    "column":
                    [
                        // 送信先テーブルのフィールドのリスト。
                        "bigint_arr",
                        "double_arr",
                        "text_arr",
                        "jsonb_obj",
                        "json_obj"
                    ],
                    "pgType":
                    {
                        // 特殊な型の設定。キーは送信先テーブルのフィールド名、値はフィールド型です。
                        "bigint_arr": "bigint[]",
                        "double_arr": "double[]",
                        "text_arr": "text[]",
                        "jsonb_obj": "jsonb",
                        "json_obj": "json"
                    }

                }
            }
        }]
    }
}

PostgreSQL SSL 認証ファイル

DataWorks で PostgreSQL データソースの接続を作成または変更する際、SSL 認証を構成できます。以下の表に、SSL 認証に関連する設定項目を示します。

PostgreSQL SSL 認証ファイルの取得

このセクションでは、ApsaraDB RDS for PostgreSQL インスタンスを例として、SSL 認証証明書を生成する方法を説明します。

1. Truststore ファイル を取得します。

   Truststore ファイル の取得方法については、「クラウド証明書を使用した SSL 暗号化の迅速な有効化」をご参照ください。

   1. RDS インスタンスページに移動し、対象リージョンの RDS インスタンスを検索して、インスタンス ID をクリックしてインスタンスの詳細ページに移動します。

   2. 保護する接続文字列を選択します (以下の図を参照)。

   

   説明

   • パブリックエンドポイントが有効になっている場合、内部エンドポイントとパブリックエンドポイントの両方が表示されます。クラウド証明書は、1 つのエンドポイントのみを保護できます。内部エンドポイントは VPC 内でより安全ですが、パブリックエンドポイントはインターネットに公開されているため、保護することを推奨します。内部エンドポイントおよびパブリックエンドポイントの表示方法については、「内部エンドポイントおよびパブリックエンドポイントの表示」をご参照ください。

   • 内部エンドポイントおよびパブリックエンドポイントの両方を保護するには、「カスタム証明書を使用した SSL 暗号化の有効化」をご参照ください。

   • クラウド証明書を構成すると、インスタンスの 実行中のステータス が SSL の変更中 に変わります。この処理には約 3 分かかります。ステータスが 操作中 に変わるまで待ってから、次の手順に進んでください。

   CA 証明書のダウンロード をクリックして、Truststore ファイル を取得します。

   

   ダウンロードされた CA 証明書パッケージには 3 つのファイルが含まれています。DataWorks で PostgreSQL データソースを構成する際は、拡張子が .pem または .p7b のファイルを Truststore ファイル フィールドにアップロードします。

2. Keystore ファイル、秘密鍵ファイル、および 秘密鍵パスワード の取得および構成

   前提条件：「クラウド証明書を使用した SSL 暗号化の迅速な有効化」または「カスタム証明書を使用した SSL 暗号化の有効化」の手順を完了し、OpenSSL ツールがインストールされている必要があります。

   説明

   Linux システムには OpenSSL ツールがデフォルトで含まれています。Windows システムを使用する場合は、「OpenSSL パッケージ」を入手してインストールする必要があります。

   Keystore ファイル、秘密鍵ファイル、および 秘密鍵パスワード の取得および構成方法については、「クライアント CA 証明書の構成」をご参照ください。

   1. OpenSSL ツールを使用して、自己署名証明書 (ca1.crt) および秘密鍵 (ca1.key) を生成します。

      openssl req -new -x509 -days 3650 -nodes -out ca1.crt -keyout ca1.key -subj "/CN=root-ca1"

   2. クライアント証明書署名要求 (CSR) ファイル (client.csr) およびクライアント秘密鍵 (client.key) を生成します。

      openssl req -new -nodes -text -out client.csr -keyout client.key -subj "/CN=<クライアントユーザー名>"

      このコマンドでは、-subj パラメーターの後の Common Name (CN) 値を、クライアントがデータベースにアクセスするために使用するユーザー名に設定します。

   3. クライアント証明書 (client.crt) を生成します。

      openssl x509 -req -in client.csr -text -days 365  -CA ca1.crt -CAkey ca1.key -CAcreateserial  -out client.crt

   4. ApsaraDB RDS for PostgreSQL サーバーがクライアント CA 証明書を検証する必要がある場合、生成された自己署名証明書ファイル ca1.crt を開きます。証明書の内容をコピーして、クライアント証明機関からの公開鍵の入力 ダイアログボックスに貼り付けます。

   5. RDS インスタンスでクライアント CA 証明書を構成した後、クライアント秘密鍵ファイル client.key を PKCS#8 形式 (client.pk8) に変換する必要があります。DataWorks で PostgreSQL データソースを構成する際は、client.pk8 ファイルを 秘密鍵ファイル フィールドにアップロードします。

      cp client.key client.pk8

   6. 秘密鍵パスワードを構成します。

      openssl pkcs8 -topk8 -inform PEM -in client.key -outform der -out client.pk8 -v1 PBE-MD5-DES

      説明

      秘密鍵パスワードを構成するコマンドを実行すると、パスワードの入力を求められます。パスワードを設定した場合、DataWorks の PostgreSQL データソース構成の「秘密鍵パスワード」フィールドに同じパスワードを使用する必要があります。

PostgreSQL SSL 認証ファイルの構成

DataWorks の PostgreSQL データソース構成に証明書ファイルをアップロードする際は、以下の手順に従います。

• Truststore ファイル：「Truststore 証明書ファイルの取得」の手順で取得した .pem または .p7b ファイルをアップロードします。

• Keystore ファイル：「クライアント証明書の生成」の手順で取得した client.crt ファイルをアップロードします。

• 秘密鍵ファイル：「秘密鍵ファイルの変換」の手順で取得した client.pk8 ファイルをアップロードします。

• 秘密鍵パスワード：「秘密鍵パスワードの構成」の手順で構成したパスワードです。

ACL の構成：RDS インスタンスページに移動し、対象リージョンの RDS インスタンスを検索して、インスタンス ID をクリックしてインスタンスの詳細ページに移動します。データセキュリティ > ACL の構成 をクリックして設定を変更します。さまざまな SSL 認証方法を選択できます。詳細については、「クライアントに SSL 接続の使用を強制する」をご参照ください。