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 or RENAME COLUMN などの他の操作 (ALTER COLUMN の動作) と一緒に使用すると、データ同期タスクは失敗します。

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

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

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

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

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

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

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

準備 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 分かかります。続行する前に、ステータスが [実行中] に変わるまで待ちます。

   [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` を開きます。証明書の内容をコピーし、[クライアント証明書機関の公開鍵を入力] ダイアログボックスに貼り付けます。

   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 データソース設定にアップロードする際は、次の手順に従います。

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

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

• [秘密鍵ファイル]: 「秘密鍵ファイルの変換」ステップで取得した `client.pk8` ファイルをアップロードします。

• [秘密鍵パスワード]: 「秘密鍵パスワードの設定」ステップで設定したパスワード。

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