Realtime Compute for Apache Flink:MongoDB カタログの管理

背景情報

このトピックでは、次の操作について説明します。

• MongoDB カタログの作成

• MongoDB カタログの表示

• MongoDB カタログの使用

• MongoDB カタログの削除

事前準備

MongoDB カタログを使用する前に、次のことを確認してください。

• ご利用の Realtime Compute for Apache Flink ワークスペースが VVR 8.0.5 以降を使用していること。`CREATE TABLE AS` 文または `CREATE DATABASE AS` 文を使用する場合は、VVR 8.0.6 以降が必要です。

制限事項

• VVR 8.0.5 以降を搭載した Realtime Compute for Apache Flink のみ、MongoDB カタログをサポートします。

• DDL 文を使用して既存の MongoDB カタログを変更することはできません。接続設定を更新するには、カタログを削除してから再作成してください。

• MongoDB カタログは読み取りアクセスのみをサポートします。MongoDB カタログを介してデータベースやテーブルを作成、変更、削除することはできません。

MongoDB カタログの作成

1. SQL エディタページの [スクリプト] タブのコードエディタに、MongoDB カタログを作成するための文を入力します。

   CREATE CATALOG <yourcatalogname> WITH(
    'type'='mongodb',
    'default-database'='<dbName>',
    'hosts'='<hosts>',
    'scheme'='<scheme>',
    'username'='<username>',
    'password'='<password>',
    'connection.options'='<connectionOptions>',
    'max.fetch.records'='100',
    'scan.flatten-nested-columns.enabled'='<flattenNestedColumns>',
    'scan.primitive-as-string'='<primitiveAsString>'
   );

   次の表にパラメーターを示します。

   パラメーター	データ型	説明	必須	デフォルト値	備考
   yourcatalogname	STRING	MongoDB カタログの名前。	はい	N/A	カスタム名を入力します。
   type	STRING	カタログのタイプ。	はい	N/A	値を mongodb に設定します。
   hosts	STRING	MongoDB インスタンスのホスト名。	はい	N/A	複数のホスト名はカンマ (,) で区切ります。
   default-database	STRING	デフォルトの MongoDB データベースの名前。	はい	N/A	N/A
   scheme	STRING	MongoDB データベースの接続プロトコル。	いいえ	mongodb	有効な値： mongodb (標準の MongoDB プロトコルを使用) mongodb+srv (DNS SRV レコードプロトコルを使用)
   username	STRING	MongoDB データベースに接続するためのユーザー名。	いいえ	N/A	MongoDB データベースで認証が有効になっている場合に必要です。
   password	STRING	MongoDB データベースに接続するためのパスワード。	いいえ	N/A	MongoDB データベースで認証が有効になっている場合に必要です。認証情報の漏洩を防ぐため、プレーンテキストのパスワードを指定する代わりにキー管理機能を使用してください。「変数とキーの管理」をご参照ください。
   connection.options	STRING	MongoDB データベースの追加の接続パラメーター。	いいえ	N/A	key=value 形式のキーと値のペアとして指定し、アンパサンド (&) で区切ります。例：connectTimeoutMS=12000&socketTimeoutMS=13000。
   max.fetch.records	INT	コレクションスキーマを推論する際に MongoDB カタログが読み取ろうとするドキュメントの最大数。	いいえ	100	N/A
   scan.flatten-nested-columns.enabled	BOOLEAN	BSON 形式のドキュメントを解析する際に、ネストされた列を再帰的に展開するかどうかを指定します。	いいえ	false	有効な値： true：ネストされた列が展開されます。列のパスが列名になります。例：{"nested": {"col": true}} は nested.col になります。 false：ネストされたドキュメントは STRING として解析されます。 重要 このパラメーターは、Flink SQL デプロイメントでソーステーブルとして使用される MongoDB カタログテーブルにのみ適用されます。
   scan.primitive-as-string	BOOLEAN	BSON 形式のドキュメントを解析する際に、すべての基本データ型を STRING として推論するかどうかを指定します。	いいえ	false	有効な値： true：すべての基本データ型が STRING として推論されます。 false：型は BSON から Flink への型マッピングから推論されます。型マッピングについては、「スキーマ推論の説明」をご参照ください。

2. カタログを作成するためのコードを選択し、コードエディタの左側にある [実行] をクリックします。

   

3. カタログリストページの左側にある [カタログ] ペインで、新しいカタログが表示されていることを確認します。

MongoDB カタログの表示

1. SQL エディタページの [スクリプト] タブのコードエディタに、次の文を入力します。

   DESCRIBE `${catalog_name}`.`${db_name}`.`${collection_name}`;

   パラメーター	説明
   ${catalog_name}	MongoDB カタログの名前。
   ${db_name}	ApsaraDB for MongoDB データベースの名前。
   ${collection_name}	ApsaraDB for MongoDB コレクションの名前。

2. コードを選択し、コードエディタの左側にある [実行] をクリックします。

   文が実行されると、結果パネルにテーブルの詳細が表示されます。

   

MongoDB カタログの使用

INSERT INTO ${other_sink_table}
SELECT...
FROM `${mongodb_catalog}`.`${db_name}`.`${collection_name}`
/*+OPTIONS('scan.incremental.snapshot.enabled'='true')*/;

CREATE TABLE IF NOT EXISTS `${target_table_name}`
WITH(...)
AS TABLE `${mongodb_catalog}`.`${db_name}`.`${collection_name}`
/*+ OPTIONS('scan.incremental.snapshot.enabled'='true', 'scan.full-changelog'='true') */;

BEGIN STATEMENT SET;

CREATE TABLE IF NOT EXISTS `some_catalog`.`some_database`.`some_table0`
AS TABLE `mongodb-catalog`.`database`.`collection0`
/*+ OPTIONS('scan.incremental.snapshot.enabled'='true', 'scan.full-changelog'='true') */;

CREATE TABLE IF NOT EXISTS `some_catalog`.`some_database`.`some_table1`
AS TABLE `mongodb-catalog`.`database`.`collection1`
/*+ OPTIONS('scan.incremental.snapshot.enabled'='true', 'scan.full-changelog'='true') */;

CREATE TABLE IF NOT EXISTS `some_catalog`.`some_database`.`some_table2`
AS TABLE `mongodb-catalog`.`database`.`collection2`
/*+ OPTIONS('scan.incremental.snapshot.enabled'='true', 'scan.full-changelog'='true') */;

END;

CREATE DATABASE IF NOT EXISTS `some_catalog`.`some_database`
AS DATABASE `mongodb-catalog`.`database`
/*+ OPTIONS('scan.incremental.snapshot.enabled'='true', 'scan.full-changelog'='true') */;

INSERT INTO ${other_sink_table}
SELECT ...
FROM ${other_source_table} AS e
JOIN `${mysql_catalog}`.`${db_name}`.`${table_name}` FOR SYSTEM_TIME AS OF e.proctime AS w
ON e.id = w.id;

INSERT INTO `${mysql_catalog}`.`${db_name}`.`${table_name}`
SELECT ...
FROM ${other_source_table}

MongoDB カタログの削除

1. SQL エディタページの [スクリプト] タブのコードエディタに、次の文を入力します。

   DROP CATALOG ${catalog_name};

   ${catalog_name} は、削除したい MongoDB カタログの名前を指定します。

2. 文を右クリックし、ショートカットメニューから [実行] を選択してカタログを削除します。

3. カタログリストページの左側にある [カタログ] ペインで、カタログが表示されなくなったことを確認します。

スキーマ推論の説明

MongoDB カタログがテーブルのスキーマを推論する際、デフォルトのパラメーターとプライマリキー情報が自動的に追加されます。カタログが BSON 形式のドキュメントを解析してコレクションのスキーマを取得する際、max.fetch.records パラメーターで指定された数のデータレコードまで読み取りを試みます。カタログは各レコードのスキーマを解析し、解析されたすべてのスキーマを単一のコレクションスキーマにマージします。

コレクションスキーマは、次の部分で構成されます。

• 物理列：MongoDB カタログは、BSON 形式のドキュメントから物理列を推論します。

• デフォルトのプライマリキー：MongoDB カタログ内のどのテーブルでも、データの重複を防ぐために _id 列がプライマリキーとして機能します。

関連ドキュメント

• MongoDB コネクタの使用方法の詳細については、「MongoDB コネクタ」をご参照ください。

• Realtime Compute for Apache Flink の組み込みカタログが要件を満たさない場合は、カスタムカタログを使用できます。詳細については、「カスタムカタログの管理」をご参照ください。