Openflow Connector for SQL Server の設定¶

前提条件¶

コネクタを設定する前に、以下の前提条件を満たしていることを確認してください。

1. Openflow Connector for SQL Server について を確認してください。

2. サポートされている SQL Serverバージョン を確認してください。

3. ランタイムの展開を設定していることを確認してください。詳細については、次のトピックをご参照ください。

   • Openflowの設定 - BYOC

   • Openflowの設定 - Snowflakeデプロイメント。

4. |OFSFSPCS-plural|を使用する場合、 必要なドメインの構成 を精査し、 SQL サーバー コネクタに必要なドメインへのアクセス許可を付与していることを確認してください。

SQL Serverインスタンスの設定¶

コネクタを設定する前に、SQL Server環境で次のタスクを実行してください。

1. 次のSQL Server の例に示すように、複製する予定の データベース と テーブル で変更の追跡を有効にします。

   CopyExpand

   ALTER DATABASE <database>
     SET CHANGE_TRACKING = ON
     (CHANGE_RETENTION = 2 DAYS, AUTO_CLEANUP = ON);
   
   ALTER TABLE <schema>.<table>
     ENABLE CHANGE_TRACKING;
   

   Show lessSee more

   Scroll to top

   注釈

   複製する予定のすべてのデータベースとテーブルに対してこれらのコマンドを実行します。

   コネクタは、複製を開始する前に、データベースとテーブルで変更追跡を有効にする必要があります。複製を予定しているすべてのテーブルで、変更の追跡が有効になっていることを確認してください。コネクタの実行中に、追加のテーブルの変更追跡を有効にすることもできます。

2. Create a login for the SQL Serverインスタンスのログインを作成する:

   CopyExpand

   CREATE LOGIN <user_name> WITH PASSWORD = '<password>';
   

   Show lessSee more

   Scroll to top

   このログインは、複製する予定のデータベースのユーザーを作成するために使用されます。

3. 各データベースで次のSQL Serverコマンドを実行して、複製するデータベースごとにユーザーを作成します。

   CopyExpand

   USE <source_database>;
   CREATE USER <user_name> FOR LOGIN <user_name>;
   

   Show lessSee more

   Scroll to top

4. 複製する各データベースのユーザーにSELECTおよびVIEW CHANGE TRACKING権限を付与します。

   CopyExpand

   GRANT SELECT ON <database>.<schema>.<table> TO <user_name>;
   GRANT VIEW CHANGE TRACKING ON <database>.<schema>.<table> TO <user_name>;
   

   Show lessSee more

   Scroll to top

   複製する予定のすべてのテーブルに対して、各データベースでこれらのコマンドを実行します。これらの権限は、前のステップで作成した各データベースのユーザーに付与する必要があります。

5. （オプション）ユーザー定義データ型（UDDT）に対する VIEW DEFINITION 権限を付与します。

   テーブルにユーザー定義データ型（UDDT）を使用する列が含まれていて、UDDT がコネクタユーザーとは異なるユーザーによって所有されている場合は、次のSQL Serverの例に示すように、コネクタユーザーに VIEW DEFINITION 権限を付与する必要があります。

   CopyExpand

   GRANT VIEW DEFINITION TO <user_name>;
   

   Show lessSee more

   Scroll to top

   この権限がない場合、UDDT を使用する列は自動的に複製から除外されます。

6. （オプション）SSL接続を構成します。

   SSL接続を使用してSQL Server に接続する場合は、データベースサーバーのルート証明書を作成します。これはコネクタを構成するときに必要です。

Snowflake環境のセットアップ¶

Snowflake管理者は以下のタスクを実行します。

1. 複製されたデータを格納するために、Snowflake内に宛先データベースを作成します。

   CopyExpand

   CREATE DATABASE <destination_database>;
   

   Show lessSee more

   Scroll to top

2. Snowflakeの サービスユーザー を作成します。

   CopyExpand

   CREATE USER <openflow_user>
     TYPE = SERVICE
     COMMENT='Service user for automated access of Openflow';
   

   Show lessSee more

   Scroll to top

3. コネクタ用Snowflakeロールを作成し、必要な権限を付与します。

   CopyExpand

   CREATE ROLE <openflow_role>;
   GRANT ROLE <openflow_role> TO USER <openflow_user>;
   GRANT USAGE ON DATABASE <destination_database> TO ROLE <openflow_role>;
   GRANT CREATE SCHEMA ON DATABASE <destination_database> TO ROLE <openflow_role>;
   

   Show lessSee more

   Scroll to top

   このロールを使用して、コネクタのSnowflakeデータベースへのアクセスを管理します。

   宛先データベースにオブジェクトを作成するには、アクセス管理に使用するロールに、データベースに対する USAGE権限およびCREATE SCHEMA権限 を付与する必要があります。

4. コネクタ用にSnowflakeウェアハウスを作成し、必要な権限を付与します。

   CopyExpand

   CREATE WAREHOUSE <openflow_warehouse> WITH
     WAREHOUSE_SIZE = 'XSMALL'
     AUTO_SUSPEND = 300
     AUTO_RESUME = TRUE;
   GRANT USAGE, OPERATE ON WAREHOUSE <openflow_warehouse> TO ROLE <openflow_role>;
   

   Show lessSee more

   Scroll to top

   Snowflakeは、XSMALLウェアハウスサイズから始めて、複製されるテーブルの数と転送されるデータの量に応じてサイズ変更を試してみることを推奨しています。多数のテーブルを扱う場合、通常はウェアハウスサイズを大きくするよりも、マルチクラスターのウェアハウスを使用した方が適切にスケールできます。詳細については、 マルチクラスターウェアハウス をご参照ください。

5. キーペア認証用のパブリックキーとプライベートキーを設定します。

   1. 安全なキーのペア（パブリックおよび秘密）を作成します。

   2. ユーザーの秘密キーをファイルに格納して、コネクタの構成に提供します。

   3. Snowflakeユーザーに公開キーを割り当てます：

      CopyExpand

      ALTER USER <openflow_user> SET RSA_PUBLIC_KEY = 'thekey';
      

      Show lessSee more

      Scroll to top

      詳細については、 キーペア認証とキーペアローテーション をご参照ください。

コネクタをインストールする¶

コネクタをインストールするには、データエンジニアとして次を実行します。

1. Openflow概要ページに移動します。Featured connectors セクションで、 View more connectors を選択します。

2. Openflowのコネクタページでコネクタを探し、 Add to runtime を選択します。

3. Select runtime ダイアログで、Available runtimes ドロップダウンリストからランタイムを選択して Add をクリックします。

   注釈

   コネクタをインストールする前に、コネクタが取り込んだデータを格納するためのデータベースとスキーマをSnowflakeで作成したことを確認します。

4. Snowflakeアカウント認証情報でデプロイメントを認証し、Snowflakeアカウントへのランタイムアプリケーションのアクセスを許可するよう求められたられたら、 Allow を選択します。コネクタのインストールプロセスは数分で完了します。

5. Snowflakeアカウント認証情報でランタイムを認証します。

コネクタプロセスグループが追加されたOpenflowキャンバスが表示されます。

コネクタを構成する¶

コネクタを構成するには、データエンジニアとして次の手順を実行します。

1. インポートしたプロセスグループを右クリックし、 Parameters を選択します。

2. 必要なパラメーター値を入力します。

   必要なパラメーター値について詳しくは、次のセクションを参照してください。

   • SQLServerソースパラメーター:SQLServerとの接続を確立するために使用します。

   • SQLServer宛先パラメーター:Snowflakeとの接続を確立するために使用します。

   • SQLServer取り込みパラメーター:複製するテーブルを指定するために使用します。

まず、SQLServerのソースパラメーターコンテキストのパラメーターを設定し、次にSQLServerの宛先パラメーターコンテキストのパラメーターを設定します。これを完了したら、コネクタを有効にします。コネクタはSQLServerとSnowflakeの両方に接続し、実行を開始します。ただし、複製するテーブルが構成に明示的に追加されるまで、コネクタはデータを複製しません。

複製する特定のテーブルを構成するには、 SQLServer 取り込みパラメーターコンテキストを編集します。SQLServer 取り込みパラメーターコンテキストに変更を適用すると、コネクタが構成を取得し、各テーブルの複製ライフサイクルが開始します。

SQL Serverのレプリカサーバーからテーブルを複製する¶

このコネクタは、トランザクションの複製<https://learn.microsoft.com/en-us/sql/relational-databases/replication/transactional/transactional-replication>`_を使用して、プライマリサーバーまたはサブスクライバーサーバーからデータを取り込むことができます。SQL Serverレプリカに接続するコネクタを構成する前に、プライマリノードとレプリカノード間の複製が正しく機能することを確認します。トランザクションの複製の設定手順については、`チュートリアル: トランザクションの複製を構成する をご参照ください。コネクタでのデータの欠落に関する問題を調査する場合は、まずコネクタが使用するレプリカサーバーに欠落している行と変更追跡イベントが存在することを確認します。

パブリッシャーではなくサブスクライバーサーバーから読み取るようにコネクタを構成するには、:ui:`SQLServer Connection URL`パラメーターにサブスクライバーサーバーURLを指定します。

テーブル複製の再開¶

（主キーが欠落している、またはスキーマの変更がサポートされていないなどの理由で） FAILED 状態のテーブルは、自動的に再開されません。テーブルが FAILED 状態になった場合、または複製を最初から再開する必要がある場合は、次の手順に従ってテーブルを削除し、複製に再度追加します。

1. 次のいずれかの方法を使用して、複製対象からテーブルを除外します。

   • テーブルを Re-snapshot Table Exclusions パラメーターに追加して、一時的に複製対象から除外します。これは、変更することを回避する必要がある Included Table Regex によってテーブルが一致する場合に簡便な方法です。

   • 取り込みパラメーターのコンテキストで、Included Table Names からテーブルを削除するか、Included Table Regex を変更してテーブルが一致しないようにします。

2. テーブルが削除されたことを確認します。

   1. Openflowランタイムキャンバスで、プロセッサーグループを右クリックし、Controller Services を選択します。

   2. コントローラーサービスをリストしたテーブルで、Table State Store 行を見つけ、行の右側にある縦3つのドットをクリックして、View State を選択します。

   重要

   続行する前に、テーブルの状態がこのリストから完全に削除されるまで待つ必要があります。この構成変更が完了するまで続行しないでください。

3. 宛先をクリーンアップする:テーブルの状態が完全に削除されたと表示されたら、Snowflake で宛先テーブルを手動で DROP します。スナップショットフェーズ中に、コネクタは既存の宛先テーブルを上書きしないことに注意してください。テーブルがまだ存在する場合、複製は再度失敗します。オプションで、ジャーナルテーブルとストリームが不要になった場合は削除することもできます。

4. 最初のステップで行った変更を元に戻して、テーブルを再度追加します。すなわち、Re-snapshot Table Exclusions からテーブルを削除するか、Included Table Names または Included Table Regex に再度追加します。その後、コネクタはテーブルのスナップショットを再度取得します。

5. 再開を確認する:前述の指示に従って Table State Store をチェックします。テーブルの状態は、ステータス NEW で表示され、次に SNAPSHOT_REPLICATION に移行し、最後に INCREMENTAL_REPLICATION になります。

テーブルの列のサブセットを複製します。¶

コネクタは、テーブルごとに複製されるデータを構成列のサブセットにフィルターできます。主キー列は除外に関係なく常に含まれます。

列フィルターを適用するには、取り込みパラメーターコンテキストの:ui:`Column Filter JSON`パラメーターを、フィルターしたいテーブルごとに1つ、フィルターオブジェクトのJSON配列に設定します。

列は、名前または正規表現パターンによって含めたり除外したりできます。テーブルごとに単一の条件を適用することも、複数の条件を組み合わせることもできます。その場合、除外は常に包含よりも優先されます。

[
    {
        "database": "<database>" | "databasePattern": "<regex>",
        "schema": "<schema>" | "schemaPattern": "<regex>",
        "table": "<table>" | "tablePattern": "<regex>",
        "included": ["<column>", "<column>"],
        "excluded": ["<column>", "<column>"],
        "includedPattern": "<regex>",
        "excludedPattern": "<regex>"
    }
]

[
    {
        "database": "my_db",
        "schema": "dbo",
        "table": "orders",
        "included": ["account_id", "status", "created_at"]
    }
]

[
    {
        "database": "my_db",
        "schema": "dbo",
        "table": "orders",
        "excluded": ["internal_note", "debug_flag"]
    }
]

[
    {
        "database": "my_db",
        "schema": "dbo",
        "table": "contacts",
        "includedPattern": ".*_email",
        "excluded": ["admin_email"]
    }
]

[
    {
        "databasePattern": "prod_.*",
        "schema": "dbo",
        "table": "customers",
        "excluded": ["internal_note"]
    }
]

[
    {"database": "my_db", "schema": "dbo", "table": "orders", "included": ["account_id", "status"]},
    {"database": "my_db", "schema": "dbo", "table": "customers", "excludedPattern": ".*_internal"}
]

パーティショニングされたテーブルを複製する¶

このコネクタは、パーティション化されたテーブルの複製をサポートしています。SQLサーバーのパーティション化されたテーブルは、すべてのパーティションからのデータを含む単一の宛先テーブルとしてSnowflakeに複製されます。

パーティション化されたテーブルを複製するには、:ref:`label-sql_server_connector_setup_instance`で説明されているように、パーティション化されたテーブルで変更追跡が有効になっていることを確認します。

テーブルでデータ変更を追跡する¶

このコネクタは、ソーステーブルのデータの現在の状態と、各ポーリング間隔から検出された変更を複製します。このデータは、宛先テーブルと同じスキーマで作成されたジャーナルテーブルに格納されます。

ジャーナルテーブル名の形式は``<source_table_name>_JOURNAL_<timestamp>_<schema_generation>``です。ここで、``<timestamp>``はソーステーブルが複製に追加されたときのエポック秒の値であり、``<schema_generation>``はソーステーブルのスキーマ変更ごとに増加する整数です。その結果、スキーマが変更されるソーステーブルには、複数のジャーナルテーブルがあります。

複製からテーブルを削除してから再度追加すると、``<timestamp>``値が変更され、``<schema_generation>``は``1``から再度開始されます。

コネクタはジャーナルテーブルをドロップすることはありませんが、複製されたソーステーブルごとに最新のジャーナルを使用し、ジャーナル上の追加専用ストリームのみを読み取ります。ストレージを回収するには、以下を実行します。

• すべてのジャーナルテーブルをいつでも切り捨てます。

• 複製から削除されたソーステーブルに関連するジャーナルテーブルをドロップします。

• アクティブに複製されたテーブルの最新の生成ジャーナルテーブルを除いて、すべてをドロップします。

例えば、コネクタがソーステーブル orders をアクティブに複製するように設定されており、以前にテーブル customers を複製から削除した場合、以下のようなジャーナルテーブルが存在する可能性があります。この場合、 orders_5678_2 を 除いて、それらのすべてをドロップできます。

customers_1234_1
customers_1234_2
orders_5678_1
orders_5678_2

マージタスクのスケジュールを構成する¶

コネクタはウェアハウスを使用して、変更データキャプチャ（CDC）データを宛先テーブルにマージします。この操作は、 MergeSnowflakeJournalTable プロセッサーによってトリガーされます。新しい変更がない場合、または MergeSnowflakeJournalTable キューで待機する新しいフローファイルがない場合、マージはトリガーされず、ウェアハウスは自動サスペンドします。

マージタスクのスケジュールCRONパラメーターでCRON式を使用して、ウェアハウスコストを制限し、マージをスケジュールされた時間のみに制限します。MergeSnowflakeJournalTableプロセッサーに送信されるフローファイルを制限し、マージは指定された期間にのみトリガーされます。スケジュールに関する詳細は、 スケジュールストラテジー をご参照ください。

フローを実行する¶

1. プレーンを右クリックし、 Enable all Controller Services を選択します。

2. インポートしたプロセスグループを右クリックし、 Start を選択します。コネクタがデータの取り込みを開始します。