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 の先行書き込みログ (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. トラストストアファイルの取得

   トラストストアファイルの取得方法の詳細については、「クラウド証明書で SSL 暗号化を有効にする」をご参照ください。

   1. RDS インスタンスページに移動し、ターゲットリージョンで RDS インスタンスを見つけ、インスタンス ID をクリックしてインスタンス詳細ページに移動します。

   2. 次の図に示すように、保護する接続文字列を選択します。

   

   説明

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

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

   • クラウド証明書を構成した後、インスタンスの 実行中のステータスはSSL の変更中に変わります。このプロセスには約 3 分かかります。ステータスが 操作中に変わるまで待ってから続行してください。

   c. [CA 証明書のダウンロード] をクリックしてトラストストアファイルを取得します。

   

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

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

   前提条件：クラウド証明書で SSL 暗号化を有効にするまたはカスタム証明書を使用して SSL 暗号化を有効にするの手順を完了し、OpenSSL ツールがインストールされていること。

   説明

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

   キーストアファイル、秘密鍵ファイル、および秘密鍵パスワードの取得と構成方法の詳細については、「クライアント 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=<Client username>"

      このコマンドでは、-subj パラメーターの後のコモンネーム (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 を開きます。証明書の内容をコピーし、[クライアント証明機関からの公開鍵を入力] ダイアログボックスに貼り付けて、クライアント CA 証明書として使用します。

   5. RDS でクライアント CA 証明書を構成した後、クライアント秘密鍵 client.key を 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 データソース構成にアップロードする際は、次の手順に従ってください。

• トラストストアファイル：「トラストストアファイルの取得」ステップで取得した .pem または .p7b ファイルをアップロードします。

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

• 秘密鍵ファイル：「秘密鍵ファイルの変換」ステップでクライアント証明書の秘密鍵から変換された client.pk8 ファイルをアップロードします。

• 秘密鍵パスワード：「秘密鍵パスワードの構成」ステップで構成したパスワードを入力します。

ACL の構成：RDS インスタンスページに移動し、ターゲットリージョンで ApsaraDB RDS for PostgreSQL インスタンスを見つけ、インスタンス ID をクリックしてインスタンス詳細ページを開きます。 をクリックして設定を変更します。さまざまな SSL 認証方法を選択できます。詳細については、「クライアントに SSL 接続の使用を強制する」をご参照ください。