様々なデータソースからデータを取り込める統合サービス「Openflow」では、Oracleコネクタを利用することができます。
Oracleコネクタを使用することで、OracleデータベースをSnowflakeに接続し、ほぼリアルタイムでデータを複製することができます。

公式ドキュメントに手順は記載されていますが、ライセンスによる操作の違いや設定の前後関係など、初見では分かりにくい部分もありました。本記事では、実際に試した結果を詳しくまとめてご紹介します。

概要

今回のレプリケーションにおける構成は以下の通りです。

AWS RDSのOracleインスタンスから、Oracleコネクタを使ってSnowflake側にテーブルをレプリケーションします。

Oracleコネクタは、OracleのCDC（Change Data Capture）機能である XStream を利用してデータの変更を検知し、Snowflakeへ転送します。XStreamはOracle GoldenGateと同等の技術基盤を持つため、後述するライセンス要件が発生します。

1. 事前準備

1.1. 規約確認

「管理者」>「規約」を開き、「Oracleライセンス規約」の「確認」をクリックします。

開いたポップアップで「Oracleライセンス規約に同意します」にチェックを入れ、「受諾」をクリックします。

受諾後、ORGADMINで規約ページを確認すると、「Openflow for Oracle」タブが有効化されていることを確認できます。

1.2. リソース作成

Snowsightにログインし、以下のSQLを実行します。

USE ROLE ACCOUNTADMIN;

-- Openflow管理者ロールの作成
CREATE ROLE IF NOT EXISTS OPENFLOW_ADMIN;
GRANT ROLE OPENFLOW_ADMIN TO USER <openflow_user>;

-- Openflow管理者ロールへの権限付与
GRANT CREATE OPENFLOW DATA PLANE INTEGRATION ON ACCOUNT TO ROLE OPENFLOW_ADMIN;
GRANT CREATE OPENFLOW RUNTIME INTEGRATION ON ACCOUNT TO ROLE OPENFLOW_ADMIN;
GRANT CREATE COMPUTE POOL ON ACCOUNT TO ROLE OPENFLOW_ADMIN;

-- Openflow管理者ロールとセカンダリロールの設定
ALTER USER <openflow_user> SET DEFAULT_ROLE = OPENFLOW_ADMIN;
ALTER USER <openflow_user> SET DEFAULT_SECONDARY_ROLES = ('ALL');

-- データ取り込み用データベース作成
CREATE DATABASE OPENFLOW_ORACLE_DB;

-- サービスユーザ作成
CREATE USER OPENFLOW_SERVICE_USER
  TYPE = SERVICE
  COMMENT='Service user for automated access of Openflow';

-- Openflow用ロール作成
CREATE ROLE OPENFLOW_ROLE;
GRANT ROLE OPENFLOW_ROLE TO USER <openflow_user>;
GRANT ROLE OPENFLOW_ROLE TO USER OPENFLOW_SERVICE_USER;
GRANT ROLE OPENFLOW_ROLE TO ROLE SYSADMIN;
GRANT USAGE ON DATABASE OPENFLOW_ORACLE_DB TO ROLE OPENFLOW_ROLE;
GRANT CREATE SCHEMA ON DATABASE OPENFLOW_ORACLE_DB
  TO ROLE OPENFLOW_ROLE;

-- Openflow用ウェアハウス作成
CREATE WAREHOUSE OPENFLOW_WH WITH
  WAREHOUSE_SIZE = 'XSMALL'
  AUTO_SUSPEND = 300
  AUTO_RESUME = TRUE;
GRANT USAGE, OPERATE ON WAREHOUSE OPENFLOW_WH
  TO ROLE OPENFLOW_ROLE;

上記SQLにより、Openflowコネクタの設定に必要なリソース一式が作成されます。

1.3. キーペア作成

Openflowの認証に使用するキーペアを作成します。

Linux環境のターミナルから以下のコマンドを実行し、秘密鍵を生成します。

openssl genrsa 2048 | openssl pkcs8 -topk8 -inform PEM -out rsa_key.p8 -nocrypt

生成された秘密鍵（rsa_key.p8）は、第三者に漏洩しないよう安全な場所に保管してください。

以下のコマンドを実行し、公開鍵を生成します。

openssl rsa -in rsa_key.p8 -pubout -out rsa_key.pub

作成した公開鍵をサービスユーザに割り当てます。

ALTER USER OPENFLOW_SERVICE_USER SET RSA_PUBLIC_KEY = 'MIIBIj...';

1.4. 外部通信IP範囲確認

以下のSQLを実行し、Snowflakeの外部通信IP範囲を取得します。

SELECT 
    value:"ipv4_prefix"::VARCHAR AS IP_CIDR_RANGE,
    value:"effective"::TIMESTAMP AS EFFECTIVE_DATE,
    value:"expires"::TIMESTAMP AS EXPIRATION_DATE
FROM TABLE(FLATTEN(INPUT => PARSE_JSON(SYSTEM$GET_SNOWFLAKE_EGRESS_IP_RANGES())));

後の手順で使用するため、IP_CIDR_RANGE の値を控えておきます。

2. デプロイメント作成

Openflow管理者ロールを付与したユーザでSnowsightにサインインし、Openflowを起動します。

「Create a deployment」をクリックします。

デプロイのロケーションは、特別な要件がなければ「Snowflake」を選択します。

データの物理的な配置場所に制限がある場合や、自社環境内で安全にマスキングや前処理を行いたいなどセキュリティ上の要件がある場合はBYOC（Bring Your Own Cloud）を選択します。

「Create deployment」をクリックします。

STATEが「Active」となれば準備完了です。

3. ランタイム作成

「Create runtime」をクリックします。

表示されたダイアログボックスで、以下の内容を入力します。

• Deployment：上記で作成したデプロイメント
• Runtime name：任意のランタイム名
• Node type：Medium（※1）
• Min/Max nodes：1（※2）
• Execute-as role：1.2で作成したOpenflow用ロール（OPENFLOW_ROLE）

※1：ランタイムのサイズは「Medium」以上である必要があります [参考]
※2：コネクタはマルチノードのランタイムをサポートしていません [参考]

4. Oracleデータベース構成

本検証では、AWS RDSを使用します。以下の設定でOracleインスタンスを作成します。

• エディション：Oracle Enterprise Edition
• VPC/サブネット/セキュリティグループ：新規作成
• パブリックアクセス：あり
• バージョン：19c（上記以外はデフォルト）

インスタンスが作成されたら、A5:SQL Mk-2等のSQLクライアントでRDSに接続します。

ホスト名はインスタンスの「接続とセキュリティ」>「エンドポイント」で確認できます。
以降の手順は、A5:SQL Mk-2を使用することを想定しています。

アーカイブREDOログの保持期間を設定

変更データを複製に利用できるようにするため、ARCHIVELOGモードを有効にする必要があります。
RDSを使用する場合、保持期間も設定する必要があります。保持期間は、ソースデータベースの変更量とストレージ容量に基づいて決定します（以下の例では24時間）。

以下をプロシージャモードで実行します。

BEGIN
    rdsadmin.rdsadmin_util.set_configuration(
        name  => 'archivelog retention hours',
        value => '24');
END;

XStreamと補足ログを有効化

RDSの制限により ALTER SYSTEM コマンドは直接実行できないため、AWSの管理画面から設定を変更します。

RDSのメニューで「パラメータグループ」>「パラメータグループの作成」をクリックし、パラメータグループを作成します。
作成されたパラメータグループの「編集」をクリックし、パラメータを以下の通り変更して保存します。

• enable_goldengate_replication：TRUE
• streams_pool_size：2684354560

対象のインスタンスを選択し、「変更」をクリックします。「DBパラメータグループ」を上記で作成したパラメータグループに変更し、「すぐに適用」を選択して変更を適用します。パラメータの変更を有効化するため、インスタンスを再起動します。

以下をプロシージャモードで実行し、データベースに補足ログを追加します。補足ログはXStreamが変更前後のデータを確実に取得するために必要な設定です。

BEGIN
    rdsadmin.rdsadmin_util.alter_supplemental_logging('ADD', 'ALL');
END;

XStream管理者ユーザー作成

XStreamコンポーネントの管理に必要なXStream管理者ユーザーを作成します。

-- XStream管理者ユーザー用の表領域作成
CREATE TABLESPACE xstream_adm_tbs 
   DATAFILE SIZE 25M AUTOEXTEND ON MAXSIZE UNLIMITED;

-- XStream管理者ユーザー作成
CREATE USER xstreamadmin IDENTIFIED BY "YOUR_XSTREAM_ADMIN_PASSWORD"
   DEFAULT TABLESPACE xstream_adm_tbs
   QUOTA UNLIMITED ON xstream_adm_tbs;

YOUR_XSTREAM_ADMIN_PASSWORD には任意のパスワードを設定します。

XStream管理者権限を付与

XStream管理者ユーザーに必要な権限を付与します。

GRANT CREATE SESSION TO xstreamadmin;

以下はプロシージャモードで実行します。

BEGIN
  DBMS_XSTREAM_AUTH.GRANT_ADMIN_PRIVILEGE(
    grantee                 => 'xstreamadmin',
    privilege_type          => 'CAPTURE',
    grant_select_privileges => TRUE);
END;

XStreamサーバーの接続ユーザーを設定

コネクタで使用するXStreamアウトバウンドサーバー接続用ユーザーを作成します。

-- 接続ユーザー作成
CREATE USER connectuser IDENTIFIED BY "YOUR_CAPTURE_USER_PASSWORD";

-- 接続ユーザーへの権限付与
GRANT CREATE SESSION, SELECT_CATALOG_ROLE, SELECT ANY TABLE TO connectuser;

XStreamアウトバウンドサーバーを作成

以下をプロシージャモードで実行し、XStreamアウトバウンドサーバーを作成します。

DECLARE
    tables  DBMS_UTILITY.UNCL_ARRAY;
    schemas DBMS_UTILITY.UNCL_ARRAY;
BEGIN
   -- 全スキーマの全テーブルを複製する場合
   tables(1) := NULL;
   schemas(1) := NULL;
   DBMS_XSTREAM_ADM.CREATE_OUTBOUND(
       server_name => 'XOUT1',
       table_names => tables,
       schema_names => schemas,
       include_ddl => TRUE
   );
END;

XStreamアウトバウンドサーバー接続ユーザーを設定

以下をプロシージャモードで実行し、XStreamアウトバウンドサーバーで使用する接続ユーザーを設定します。

BEGIN
    DBMS_XSTREAM_ADM.ALTER_OUTBOUND(
        server_name  => 'XOUT1',
        connect_user => 'connectuser');
END;

取り込み対象テーブル作成

テスト用ユーザー（スキーマ）を作成します。

-- テスト用ユーザー（スキーマ）の作成と権限付与
CREATE USER OPENFLOW_TEST IDENTIFIED BY password123
    DEFAULT TABLESPACE USERS
    QUOTA UNLIMITED ON USERS;
GRANT CREATE SESSION, CREATE TABLE TO OPENFLOW_TEST;

OPENFLOW_TESTユーザーに切り替え、テーブルとテストデータを作成します。

-- テーブル作成
CREATE TABLE ORACLE_SYNC_SAMPLE (
    ID NUMBER PRIMARY KEY,
    CONTENT VARCHAR2(100),
    CREATED_AT TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- テストデータ挿入
INSERT ALL
INTO ORACLE_SYNC_SAMPLE (ID, CONTENT) VALUES (1, 'aaa')
INTO ORACLE_SYNC_SAMPLE (ID, CONTENT) VALUES (2, 'bbb')
SELECT * FROM DUAL;

-- データ確認
SELECT * FROM ORACLE_SYNC_SAMPLE;

5. 通信許可設定

セキュリティグループ更新

SnowflakeからOracleへの通信を許可する設定を行います。

AWS RDSのVPCセキュリティグループに、「1.4. 外部通信IP範囲確認」で取得したSnowflakeの外部通信IP範囲からの内向き通信を許可するルールを追加します。

• プロトコル：TCP
• ポート範囲：1521
• ソース：Snowflakeの外部通信IP範囲

接続エラーが発生した場合は、まずこのセキュリティグループの設定を確認してください。

許可ドメイン構成

OracleからSnowflakeへの通信を許可する設定を行います。

以下のSQLで、ネットワークルールと外部アクセス統合を作成します。

-- Oracleインスタンスへの通信を定義するネットワークルール
CREATE OR REPLACE NETWORK RULE RDS_NETWORK_RULE
  TYPE = HOST_PORT
  MODE = EGRESS
  VALUE_LIST = ('<AWS RDSエンドポイント>:1521');

-- ネットワークルールを紐付けた外部アクセス統合（EAI）の作成
CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION RDS_OPENFLOW_EAI
  ALLOWED_NETWORK_RULES = (RDS_NETWORK_RULE)
  ENABLED = TRUE
  COMMENT = 'External Access Integration for Openflow connectivity';

-- 外部アクセス統合へのアクセス権付与
GRANT USAGE ON INTEGRATION RDS_OPENFLOW_EAI TO ROLE OPENFLOW_ROLE;
GRANT USAGE ON INTEGRATION RDS_OPENFLOW_EAI TO ROLE OPENFLOW_ADMIN;

OpenflowのRuntimeタブを開き、対象のランタイムのメニューから「External access integrations」を選択し、上記で作成した外部アクセス統合にチェックを入れて保存します。

6. コネクタ構成

「Featured connectors」の「View more connectors」をクリックします。

コネクタページでOracleコネクタを探し、「Install」をクリックします。
今回はBYOL（独立ライセンス）のため「Oracle Independent License」を選択します。

表示されたポップアップでランタイムを選択し、「Add」をクリックします。
アプリケーションのアクセス要求画面が開いたら、「許可」をクリックします。

インストール処理が完了すると、Openflowキャンバス画面が表示されます。

ランタイムを右クリックし、「Parameters」を選択します。

以下のパラメータ値を入力します。

• Oracle Destination Parameters
  • Destination Database：レプリケーション先データベース（OPENFLOWORACLEDB）
  • Snowflake Account Identifier：アカウント識別子（[組織名]-[アカウント名]）
  • Snowflake Private Key：キーペア認証で使用する秘密鍵
  • Snowflake Role：サービスユーザーに付与されたロール（OPENFLOWROLE）
  • Snowflake Username：インスタンスへの接続に使用するユーザー名（OPENFLOWSERVICEUSER）
  • Snowflake Warehouse：クエリ実行に使用するウェアハウス（OPENFLOWWH）
• Oracle Ingestion Parameters
  • Included Table Names：取り込み対象のテーブルパス（ORCL.OPENFLOWTEST.ORACLESYNC_SAMPLE）
  • Included Table Regex：取り込み対象のテーブルパスに一致する正規表現（ORCL\.OPENFLOW_TEST\..*）
• Oracle Source Parameters
  • Oracle Connection URL：データベース接続用JDBC URL（jdbc:oracle:thin:@//<AWS RDSエンドポイント>:1521/ORCL）
  • Oracle Password：XStreamサーバーにアクセスできる接続ユーザーのパスワード（connectuserのパスワード）
  • Oracle Username：XStreamサーバーにアクセスできる接続ユーザーのユーザー名（connectuser）
  • XStream Out Server Name：XStreamサーバーの名前（XOUT1）
  • XStream Out Server URL：XStreamのデータベース接続のJDBC URL（jdbc:oracle:oci:@//<AWS RDSエンドポイント>:1521/ORCL）

上記以外のパラメータはデフォルトのままで問題ありません。

以上で、レプリケーションの設定は完了です。

キャンバス画面に戻ってランタイムを右クリックし、「Enable all Controller Services」をクリック後、「Start」をクリックします。
取り込みが開始すると、以下のように処理状況が表示されます。

データベースを開くと、取り込み先データベースにOracleと同様のスキーマとテーブルが作成されていることが確認できます。
スキーマは「（Oracle側のDB名）_（Oracle側のスキーマ名）」の形式で作成されます。

テーブルには元のカラムに加え、レプリケーション時のメタデータ（タイムスタンプ、元データの削除有無）が追加されています。

注意事項

便利なOracleコネクタですが、いくつか注意点があります。

ライセンスに関する注意

OracleコネクタはOracleのXStream（CDCのしくみ）を利用しており、ライセンスの種類に応じて必要な対応が異なります。
ライセンス形態は以下の2つです。

• 組み込みライセンス
  • Oracle XStreamのライセンスをSnowflake経由で直接購入する場合
  • Snowflakeからライセンス費用が請求される
  • 60日間のトライアル期間を過ぎると3年分のコミットメントが確定する（途中でキャンセル不可）
• 独立ライセンス（BYOL）
  • Oracle GoldenGate（または相当するXStream資格）のライセンスを保持している場合
  • Snowflakeからライセンス費用が請求されない
  • ライセンス準拠の責任はユーザー側が負う

既にOracle GoldenGateのライセンスを保有している場合は独立ライセンスを選択できます。そうでない場合は組み込みライセンスになりますが、60日間のトライアル期間を過ぎると3年分の請求が確定してしまう点に注意が必要です。トライアルで評価後にやめる場合は、60日以内のキャンセルを忘れないようにしましょう。

課金に関する注意

Openflowでランタイムを起動したまま放置すると、コネクタが稼働していなくても課金が発生します（目安として日次で4〜5クレジット程度）。使用しない場合はランタイムのメニューから「Suspend」で停止するようにしましょう。

1日5クレジット、Enterpriseエディションの場合で試算すると、1か月あたりのコストは以下の通りです。

5クレジット × 30日 × $4.30/クレジット × 160円/ドル ≒ 103,200円/月

これにライセンス費用も加算されるため、ランニングコストは事前に十分確認しておくことをお勧めします。

データ型マッピング

OracleからSnowflakeへのデータ型マッピングは公式ドキュメントに記載の通りです。
NUMBERの精度が未定義の場合に NUMBER(38, 19) にマッピングされるなど、注意が必要な仕様があります。

さいごに

Openflow Connector for Oracleについて、詳しくご紹介しました。

OracleデータベースからSnowflakeへほぼリアルタイムでデータをレプリケーションできる便利な機能ですが、ライセンスや課金など留意すべき点もあります。

特定テーブルのレプリケーションが目的であれば、こちらの記事でご紹介しているようにLambdaを使う安価な方法もあります。ユースケースに応じて最適な方法を選択してください。

参考

• Openflowについて
• Openflow Connector for Oracle について