Openflow Connector for SQL Server を設定する

注釈

コネクタには、 コネクタ利用規約 が適用されます。

このトピックでは、 Openflow Connector for SQL Server を設定する手順について説明します。

前提条件

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

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

  3. Openflowを設定した ことを確認します。

  4. データベース管理者として、以下のタスクを実行します。

    1. データベーステーブル の変更追跡を有効にします。コネクタでは、複製を開始する前にデータベースとテーブルで変更追跡が有効になっている必要があります。複製するすべてのテーブルで変更追跡が有効になっていることを確認してください。コネクタの実行中に、さらに多くのテーブルの変更追跡を有効にすることもできます。以下のコードスニペットをご参照ください。

      ALTER DATABASE <database>
      SET CHANGE_TRACKING = ON
      (CHANGE_RETENTION = 2 DAYS, AUTO_CLEANUP = ON);
      
      ALTER TABLE <schema>.<table>
      ENABLE CHANGE_TRACKING
      WITH (TRACK_COLUMNS_UPDATED = ON);
      
      Copy
    2. コネクタのユーザーを作成します。コネクタでは、複製テーブルに対する VIEW CHANGE TRACKING 権限が付与されたユーザーが必要です。そのユーザーに、コネクタの構成にアクセスするためのパスワードを与えます。

      CREATE LOGIN <user_name> WITH PASSWORD = <password>;
      CREATE USER <user_name> FOR LOGIN <user_name>;
      GRANT SELECT ON <schema>.<table> TO <user_name>;
      GRANT VIEW CHANGE TRACKING ON <schema>.<table> TO <user_name>;
      
      Copy
    3. SSL 経由で接続します。SQL Serverへの SSL 接続を使用する場合は、データベースサーバーのルート証明書を準備します。これは、構成時に必要です。

  5. Snowflakeアカウント管理者として、以下のタスクを実行します。

    1. タイプを SERVICE としてSnowflakeユーザーを作成します。複製データを格納するデータベースを作成し、Snowflakeユーザーに USAGE および CREATE SCHEMA 権限 を付与して、そのデータベースにオブジェクトを作成する権限を設定します。

      CREATE DATABASE <destination_database>;
      CREATE USER <openflow_user> TYPE=SERVICE COMMENT='Service user for automated access of Openflow';
      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>;
      CREATE WAREHOUSE <openflow_warehouse>
           WITH
               WAREHOUSE_SIZE = 'MEDIUM'
               AUTO_SUSPEND = 300
               AUTO_RESUME = TRUE;
      GRANT USAGE, OPERATE ON WAREHOUSE <openflow_warehouse> TO ROLE <openflow_role>;
      
      Copy
    2. セキュアキーのペア(公開キーと秘密キー)を作成します。ユーザーの秘密キーをファイルに格納して、コネクタの構成に提供します。Snowflakeサービスユーザーに公開キーを割り当てます:

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

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

    3. コネクタが使用するウェアハウスを指定します。まずは MEDIUM のウェアハウスサイズから始め、複製するテーブルの量や転送するデータ量に応じてサイズを試してみてください。テーブル数が大きい場合は、通常、ウェアハウスのサイズよりも、 マルチクラスターウェアハウス を使用した方がスケーリングが向上します。

コネクタを設定する

データエンジニアとして、以下のタスクを実行してコネクタを構成します。

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

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

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

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

  4. Add を選択します。

    注釈

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

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

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

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

コネクタを構成する

以下のユースケースにコネクタを構成できます。

リアルタイムでテーブルのセットを複製する

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

  2. フローパラメーター の説明に従って、必要なパラメーター値を入力します。

フローパラメーター

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

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

SQLServer ソースパラメーターコンテキスト

パラメーター

説明

SQL Server接続 URL

ソースデータベースへの完全な JDBC URL。

例:

  • jdbc:sqlserver://example.com:1433;encrypt=false;databaseName=<example_database>

SQL Server JDBC ドライバー

SQL Server JDBC ドライバー をアップロードするには、 Reference asset チェックボックスを選択します。

SQL Server SSL モード

SSL 接続を有効または無効にします。

SQL Serverルート SSL 証明書

データベースのルート証明書のフルコンテンツ。SSL が無効の場合はオプション。

SQL Serverユーザー名

コネクタのユーザー名。

SQL Serverパスワード

コネクタのパスワード。

SQLServer 宛先パラメーターコンテキスト

パラメーター

説明

宛先データベース

データが永続化されるデータベース。既にSnowflakeに存在している必要があります

Snowflakeアカウント識別子

データが永続化されるSnowflakeアカウント名(形式: [organization-name]- [account-name])。

Snowflake認証ストラテジー

Snowflakeへの認証のストラテジー。可能な値: SPCS でフローを実行している場合は SNOWFLAKE_SESSION_TOKEN、秘密キーを使用してアクセスを設定する場合は KEY_PAIR

Snowflake秘密キー

認証に使用される RSA 秘密キー。RSA キーは、 PKCS8 標準に従って書式設定され、標準的な PEM ヘッダーとフッターを持つ必要があります。Snowflake秘密キーファイルまたはSnowflake秘密キーのいずれかを定義する必要があることに注意してください。

Snowflake秘密キーファイル

Snowflakeへの認証に使用される RSA 秘密キーを含むファイル。 PKCS8 標準に従って書式設定され、標準的な PEM ヘッダーとフッターを持ちます。ヘッダー行は -----BEGIN PRIVATE で始まります。 Reference asset チェックボックスを選択し、秘密キーファイルをアップロードします。

Snowflake秘密キーパスワード

Snowflake秘密キーファイルに関連付けられたパスワード

Snowflakeロール

クエリ実行時に使用されるSnowflakeロール

Snowflakeのユーザー名

Snowflakeインスタンスへの接続に使用するユーザー名

Snowflakeウェアハウス

クエリの実行に使用されるSnowflakeウェアハウス

SQLServer 取り込みパラメーターコンテキスト

パラメーター

説明

含まれるテーブル名

データベースとスキーマを含むテーブルパスのコンマ区切りリスト。例: database_public.public.my_table, other_database.other_schema.other_table

含まれるテーブル正規表現

テーブルパスに一致させる正規表現。式に一致するすべてのパスが複製され、後から作成されたパターンに一致する新しいテーブルも自動的に含まれます。例: database_public.public\.auto_.*

フィルター JSON

完全修飾されたテーブル名のリストと、複製に含めるべき列名の正規表現パターンを含む JSON。例: [ {"database":"database_public","schema":"public", "table":"table1", "includedPattern":".*name"} ] は、 database_public データベースと public スキーマから、 table1name で終わるすべての列を含めます。

タスクスケジュール CRON をマージする

ジャーナルから宛先テーブルへのマージ操作がトリガーされる期間を定義する CRON 式。連続的なマージやタイムスケジュールでウェアハウスの実行時間を制限したい場合は、 * * * * * ? に設定します。

例:

  • * 0 * * * ? という文字列は、丸1時間で1分間のマージをスケジュールしたいことを示しています。

  • * 20 14 ? * MON-FRI という文字列は、毎週月曜日から金曜日の2:20 PM にマージをスケジュールしたいことを示しています。

その他の情報と例については、 Quartzドキュメント のcronトリガーチュートリアルをご参照ください。

テーブルを削除し、複製に再追加する

複製からテーブルを削除するには、複製パラメーターコンテキストの Included Table Names または Included Table Regex パラメーターからテーブルが削除されていることを確認します。

後でテーブルを複製に再追加する場合は、まずSnowflakeで対応する宛先テーブルを削除します。その後、 Included Table Names または Included Table Regex パラメーターにテーブルを追加して戻します。これにより、テーブルの複製プロセスが新しく開始されます。

このアプローチは、失敗したテーブル複製シナリオからの復旧にも使用できます。

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

コネクタは、テーブルごとに複製されるデータを構成列のサブセットにフィルターできます。

列にフィルターを適用するには、複製パラメーターコンテキストのColumn Filterプロパティを変更し、フィルターを適用したいテーブルごとに1エントリずつ、構成の配列を追加します。

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

以下の例は、利用可能なフィールドを示しています。 schema および table フィールドは必須です。 includedexcludedincludedPatternexcludedPattern のうち1つ以上が必要です。

[
    {
        "schema": "<source table schema>",
        "table" : "<source table name>",
        "included": ["<column name>", "<column name>"],
        "excluded": ["<column name>", "<column name>"],
        "includedPattern": "<regular expression>",
        "excludedPattern": "<regular expression>",
    }
]
Copy

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

コネクタは、ソーステーブルのデータの現在の状態だけでなく、各行の中間状態も複製します。しかし、変更追跡メカニズムが使用されているため、すべての行のすべての中間状態が複製されるという保証はありません。

このデータは、宛先テーブルと同じスキーマで作成されたジャーナルテーブルに格納されます。

ジャーナルテーブル名の形式は次のとおりです: <ソーステーブル名>_JOURNAL_<スキーマ生成>

ここで、<スキーマ生成>は、ソーステーブルのスキーマが変更されるたびに増加する整数です。つまり、スキーマが変更されたソーステーブルは、複数のジャーナルテーブルを持つことになります。

重要

Snowflakeでは、ジャーナルテーブルやその中のデータを一切変更しないことを推奨しています。これらは、複製プロセスの一環として、コネクタが宛先テーブルを更新するために使用されます。

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

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

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

フローを実行する

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

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