SLS カタログの管理方法 - Realtime Compute for Apache Flink - Alibaba Cloud ドキュメントセンター

Simple Log Service (SLS) カタログは、SLS Logstore 内のログを自動的に解析し、テーブルスキーマを推定します。SLS カタログを構成すると、Flink SQL で SLS テーブルスキーマを明示的に宣言することなく、Flink ジョブから直接 SLS Logstore にアクセスできます。これにより、フィールド情報の取得がシームレスに行えます。本トピックでは、SLS カタログの作成、表示、使用、および削除方法について説明します。

背景

本トピックでは、以下の SLS カタログ管理に関する項目を扱います：

SLS カタログの作成
SLS カタログの表示
SLS Logstore の表示
SLS カタログの使用
SLS カタログの削除

制限事項

すべてのフィールドは文字列型（String）として解析されます。他のデータの型が必要な場合は、Flink SQL を使用して変換してください。
SLS カタログは、Ververica Runtime（VVR）6.0.7 以降を使用する Flink ジョブでのみサポートされます。
DDL 文による既存の SLS カタログの変更はできません。
データベースやテーブルの作成、変更、削除はサポートされていません。クエリ実行のみ可能です。
SLS カタログが提供するテーブルは、Flink SQL ジョブにおけるソーステーブルおよび結果テーブルとして使用できますが、ルックアップ用ディメンションテーブルとしては使用できません。

注意事項

SLS カタログは、サンプルデータを解析することでテーブルスキーマを生成します。データ形式が一貫していない Logstore の場合、カタログはデフォルトですべてのカラムを保持し、可能な限り広範なスキーマを返します。Logstore のデータ形式が変更された場合、異なるタイミングで SLS カタログが取得するテーブルスキーマは不一致となる可能性があります。ジョブ再起動前後で推定されたスキーマが異なると、ジョブ実行時に問題が発生する場合があります。

たとえば、Flink SQL ジョブが SLS カタログ内のテーブルを参照している場合、一定期間実行後にセーブポイントからジョブを再起動すると、以前の実行時とは異なるスキーマが取得される可能性があります。SQL ジョブの実行計画は、以前のスキーマで生成されたバージョンに基づいています。これにより、フィルター条件やフィールド値の取得などの下流処理で不整合が生じる場合があります。そのため、固定されたスキーマを使用するには、Flink SQL ジョブ内で `CREATE TEMPORARY TABLE` を使用して SLS テーブルを作成してください。

SLS カタログの作成

SLS カタログは、UI または SQL コマンドを実行して構成できます。UI を使用することを推奨します。

UI を使用した方法

[データ管理] ページに移動します。
1. 「Realtime Compute コンソール」にログインし、対象のワークスペースの [コンソール] 列で [アクション] をクリックします。
2. [データ管理] をクリックします。
[カタログの作成] をクリックし、SLS を選択してから [次へ] をクリックします。

パラメーターを構成します。

重要

カタログを作成した後は、これらの設定を変更できません。設定を変更するには、カタログを削除して新しいカタログを作成してください。

パラメーター	型	説明	必須	備考
カタログ名	String	SLS カタログの名前です。	はい	任意の英語名を入力します。
エンドポイント	String	エンドポイントアドレスです。	はい	詳細については、「エンドポイント」をご参照ください。
プロジェクト	String	SLS プロジェクトの名前です。	はい	該当なし。
accessId	String	Alibaba Cloud アカウントの AccessKey ID です。	はい	詳細については、「AccessKey ID および AccessKey Secret の確認方法」をご参照ください。重要 AccessKey 情報を保護するため、AccessKey の値を指定するには変数を使用してください。詳細については、「プロジェクト変数」をご参照ください。
accessKey	String	Alibaba Cloud アカウントの AccessKey シークレット。	はい

[確認] をクリックします。
作成したカタログは、左側の[Metadata] エリアで表示できます。

SQL コマンド

[データクエリ] エディターで、SLS カタログを構成するコマンドを入力します。

CREATE CATALOG <catalogName> WITH(
  'type'='sls',
  'endpoint'='<brokers>',
  'project'='project',
  'accessId'='${secret_values.ak_id}',
  'accessKey'='${secret_values.ak_secret}'
)

パラメーター	型	説明	必須	備考
catalogName	String	SLS カタログの名前です。	はい	任意の英語名を入力します。
type	String	カタログのタイプ。	はい	値は固定で sls です。
endpoint	String	エンドポイントアドレスです。	はい	詳細については、「エンドポイント」をご参照ください。
project	String	SLS プロジェクトの名前です。	はい	該当なし。
accessId	String	Alibaba Cloud アカウントの AccessKey ID です。	はい	詳細については、「AccessKey ID および AccessKey Secret の確認方法」をご参照ください。重要 AccessKey 情報を保護するため、AccessKey ID の値を指定するにはシークレット管理機能を使用してください。詳細については、「変数の管理」をご参照ください。
accessKey	String	Alibaba Cloud アカウントの AccessKey シークレット。	はい	詳細については、「AccessKey ID および AccessKey Secret の確認方法」をご参照ください。重要 AccessKey 情報を保護するため、AccessKey Secret の値を指定するにはシークレット管理機能を使用してください。詳細については、「変数の管理」をご参照ください。

カタログを作成するコードを選択し、行番号の左側にある [実行] をクリックします。また、カタログを作成するコード上にカーソルを配置し、右クリックして [実行] を選択することもできます。
左側の[メタデータ] エリアで、作成されたカタログを表示します。

SLS カタログの表示

SLS カタログを構成した後、以下の手順に従って SLS メタデータを表示します。

[データ管理] ページに移動します。
1. Realtime Computing for Apache Flink コンソールにログインします。
2. 対象のワークスペースの [操作] 列で、[コンソール] をクリックします。
3. [データ管理] をクリックします。
[カタログ一覧] ページで、[名前] および [タイプ] 列を表示します。
説明
カタログ内の Logstore の情報を表示するには、[表示] をクリックします。

SLS Logstore の表示

[データクエリ] エディターで、以下のコマンドを入力します。
```
DESCRIBE `${catalog_name}`.`${project_name}`.`${logstore_name}`;
```
パラメーター
説明
${catalog_name}
SLS カタログの名前です。
${project_name}
SLS プロジェクトの名前です。
${logstore_name}
SLS Logstore の名前です。
カタログを表示するために使用されるコードを選択し、コードの左側にある行番号の横にある [実行] をクリックします。
正常に実行されると、結果にテーブルの詳細が表示されます。

SLS カタログの使用

ソーステーブルとして、SLS Logstore からデータを読み取る場合。
```
INSERT INTO ${other_sink_table}
SELECT...
FROM `${catalog_name}`.`${project_name}`.`${logstore_name}`/*+OPTIONS('startTime'='2023-06-01 00:00:00')*/;
```
説明
SLS カタログのテーブルを使用する際に追加の `WITH` パラメーターを指定するには、SQL ヒントワードを使用してパラメーターを追加します。たとえば、前述の SQL ステートメントでは、SQL ヒントワードを使用してデータ消費の開始時刻を '2023-06-01 00:00:00' に指定しています。その他のパラメーターの詳細については、「Simple Log Service コネクタ」をご参照ください。

結果テーブルとして、SLS Logstore にデータを書き込む場合。

INSERT INTO `${catalog_name}`.`${project_name}`.`${logstore_name}`
SELECT ... 
FROM ${other_source_table}

重要

ソース Logstore から読み取る場合、または宛先 Logstore に書き込む前に、Logstore のインデックスを有効化する必要があります。これは、カタログが結果テーブルからデータを読み取り、書き込むデータのスキーマが宛先 SLS Logstore のスキーマと一致することを検証するためです。インデックスの有効化方法の詳細については、「インデックスの作成」をご参照ください。

SLS カタログの削除

SLS カタログは、UI または SQL コマンドを実行して削除できます。UI を使用することを推奨します。

警告

SLS カタログを削除しても、実行中のジョブには影響しません。ただし、削除されたカタログのテーブルを使用するジョブは、公開または再起動時にテーブルが見つからないため失敗します。慎重に操作してください。

UI を使用した方法

[データ管理] ページに移動します。
1. Realtime Computing for Apache Flink コンソールにログインします。
2. 対象ワークスペースの [操作] 列で、[コンソール] をクリックします。
3. [データ管理] をクリックします。
[カタログリスト] ページで、対象のカタログ名の横の [アクション] 列にある [削除] をクリックします。
表示されるメッセージで、[削除] をクリックします。
左側の[メタデータ]エリアで、対象カタログが削除されたことを確認する。

SQL コマンド

[データクエリ] エディターで、以下のコマンドを入力します。
```
DROP CATALOG ${catalog_name};
```
${catalog_name} は、削除対象の SLS カタログの名前です。
カタログを削除するコマンドを右クリックし、[実行] を選択します。
左側の [メタデータ] エリアで、対象のカタログが削除されたかを確認します。

SLS カタログからのテーブル情報の詳細

テーブルの使いやすさを高めるため、SLS カタログは派生テーブルにデフォルトの構成パラメーターおよびメタデータを追加します。以下に、SLS カタログから取得されるテーブルの詳細を示します：

SLS テーブルのスキーマ推定
SLS カタログがログを解析してトピックのスキーマを取得する際、1 つのメッセージを消費してデータスキーマを解析します。SLS カタログは、SLS ログからフィールド名および型を解析します。SLS ではすべてのログデータが文字列として保存されるため、すべてのフィールドは文字列型（String）となります。

デフォルトのテーブルパラメーター

パラメーター	説明	備考
コネクター	コネクターのタイプ。	値は固定で sls です。
エンドポイント	エンドポイントアドレスです。	詳細については、「エンドポイント」をご参照ください。
プロジェクト	SLS プロジェクトの名前です。	該当なし。
ログストア	SLS Logstore または metricstore の名前です。	該当なし。
accessId	Alibaba Cloud アカウントの AccessKey ID です。	詳細については、「AccessKey ID および AccessKey Secret の確認方法」をご参照ください。重要 AccessKey 情報を保護するため、AccessKey ID の値を指定するにはシークレット管理機能を使用してください。詳細については、「変数の管理」をご参照ください。
accessKey	Alibaba Cloud アカウントの AccessKey シークレット。	詳細については、「AccessKey ID および AccessKey Secret の確認方法」をご参照ください。重要 AccessKey 情報を保護するため、AccessKey Secret の値を指定するにはシークレット管理機能を使用してください。詳細については、「変数の管理」をご参照ください。

パラメーター	説明
${catalog_name}	SLS カタログの名前です。
${project_name}	SLS プロジェクトの名前です。
${logstore_name}	SLS Logstore の名前です。