Flink SQL ストリーミングノードの使用方法 - DataWorks - Alibaba Cloud ドキュメントセンター

DataWorks の Flink SQL ストリーミングノードを使用すると、標準 SQL でリアルタイム処理ロジックを定義できます。このノードは使いやすく、豊富な SQL をサポートし、堅牢な状態管理とフォールトトレランスを提供します。イベント時間と処理時間の両方に対応し、柔軟にスケーリングできます。このノードは Kafka や HDFS などのシステムと簡単に統合でき、詳細なログとパフォーマンス監視ツールを提供します。まず、DataWorks プロジェクトに Flink SQL ストリーミングタスクを追加し、SQL 文を記述します。このトピックでは、DataWorks で Flink SQL ストリーミングノードを開発および使用して、リアルタイムデータを処理する方法について説明します。

前提条件

管理センターで Realtime Compute for Apache Flink のコンピューティングリソースが関連付けられている必要があります。詳細については、「コンピュートエンジンのバインド」をご参照ください。
Flink SQL ストリーミングノードが作成されている必要があります。詳細については、「スケジューリングワークフロー用のノードの作成」をご参照ください。
DataWorks が Realtime Compute for Apache Flink API を呼び出すために使用する RAM ユーザーまたは RAM ロールに、必要な OpenAPI 権限が付与されている必要があります。これらの権限により、DataWorks はノードタスクを Flink クラスターに送信してデプロイできます。詳細については、「権限の付与」をご参照ください。
```
{
  "Version": "1",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["stream:CreateDeployment", "stream:UpdateDeployment", "stream:GetDeployment", "stream:DeleteDeployment"],
      "Resource": ["*"]
    }
  ]
}
```

制限事項

このノードはワークフローでは使用できません。スタンドアロンノードとして開発および実行する必要があります。
サーバーレスリソースグループのみ使用できます。従来の専用スケジューリングリソースグループは使用できません。

ステップ 1：Flink SQL ストリーミングノードの開発

Flink SQL ストリーミングノードの編集ページで、以下の説明に従ってノードタスクを開発します。

SQL コードの開発

SQL エディタでは、${variable_name} 形式で変数を定義し、Real-Time configuration パネルの Script Parameters セクションでこれらの変数に値を割り当てることで、スケジューリングシナリオでパラメーターを動的に渡すことができます。例：

-- ソーステーブル datagen_source を作成します。
CREATE TEMPORARY TABLE datagen_source(
  name VARCHAR
) WITH (
  'connector' = 'datagen'
);

-- 結果テーブル blackhole_sink を作成します。
CREATE TEMPORARY TABLE blackhole_sink(
  name  VARCHAR
) WITH (
  'connector' = 'blackhole'
);

-- ソーステーブルから結果テーブルにデータを挿入します。
INSERT INTO blackhole_sink
SELECT
  name
FROM datagen_source WHERE LENGTH(name) > ${name_length};

説明

この例では、パラメーター name_length の値は 5 です。このパラメーターは、名前の長さが 5 文字を超えるレコードのみを処理するようにデータをフィルタリングします。

ステップ 2：Flink SQL ストリーミングノードの設定

Flink SQL ストリーミングノードタスクを次のパラメーターで設定します。

Flink リソースの設定

Real-Time configuration パネルの Flink resource information セクションで、選択した Resource Mode に基づいて以下のパラメーターを設定します。詳細については、「デプロイメントリソースを設定する」をご参照ください。

パラメーター	説明
[Flink cluster]	管理センターで関連付けられたフルマネージド Flink コンピューティングリソースの名前。
[Flink engine version]	要件に基づいて Flink エンジンバージョンを選択します。
[Resource Group]	Flink ネットワークに接続できるサーバーレスリソースグループを選択します。
Resource Mode は、次の 2 つのモードをサポートしています。詳細については、「デプロイメントリソースの設定」をご参照ください。 [Basic mode] (デフォルト): 初心者やシンプルなユースケースに最適です。このモードでは、デフォルト設定と簡易設定を使用して、Flink デプロイを迅速に開始および実行します。 [Expert mode]：経験豊富なユーザーに高度な設定オプションを提供し、パフォーマンスとリソースをきめ細かくチューニングすることで、複雑な要件や高いパフォーマンス要件を満たすことができます。選択したモードに基づいてパラメーターを設定します。Flink アーキテクチャをより深く理解することで、これらのパラメーターをより効果的に設定できます。詳細については、「Flink Architecture \| Apache Flink」をご参照ください。
[Basic mode]
JobManager CPU	Flink のベストプラクティスに基づくと、JobManager が安定して動作するには、最低 0.5 CPU コアと 2 GiB のメモリが必要です。1 CPU コアと 4 GiB のメモリを推奨します。最大 16 CPU コアまで設定できます。クラスターサイズとデプロイの複雑さに基づいて設定を調整してください。
JobManager メモリ	JobManager のメモリは、タスクのスケジューリングと管理機能に影響します。安定した効率的な動作を確保するため、2 GiB から 64 GiB の範囲を推奨します。クラスターサイズとデプロイ要件に基づいてサイズを調整してください。
TaskManager CPU	TaskManager の CPU リソースは、タスク処理能力に影響します。Flink のベストプラクティスに基づくと、最低 0.5 CPU コアと 2 GiB のメモリを推奨します。標準設定は 1 CPU コアと 4 GiB のメモリです。最大 16 CPU コアまで設定できます。実際のニーズに基づいて設定を調整してください。
TaskManager メモリ	TaskManager のメモリは、処理できるデータ量とパフォーマンスを決定します。安定した実行と効率的な処理を確保するため、メモリサイズは最低 2 GiB で、最大 64 GiB まで設定できます。
[Concurrency]	Flink デプロイ内の並列タスクの数。並列度を高くすると、処理速度とリソース使用率を向上させることができます。クラスターリソースとデプロイの特性に基づいて適切な値を設定してください。
[Number of slots per TaskManager]	TaskManager あたりのスロット数は、同時に実行できるタスクの数を決定します。この設定を調整することで、リソース使用率とデプロイの並列処理能力を最適化できます。
[Expert mode]
JobManager CPU	Flink のベストプラクティスに基づくと、JobManager が安定して動作するには、最低 0.5 CPU コアと 2 GiB のメモリが必要です。最大 16 CPU コアまで設定できます。クラスターサイズとデプロイの複雑さに基づいて設定を調整してください。
JobManager メモリ	JobManager のメモリは、タスクのスケジューリングと管理機能に影響します。安定した効率的な動作を確保するため、2 GiB から 64 GiB の範囲を推奨します。クラスターサイズとデプロイ要件に基づいてサイズを調整してください。
[Number of slots per TaskManager]	TaskManager あたりのスロット数は、同時に実行できるタスクの数を決定します。この設定を調整することで、リソース使用率とデプロイの並列処理能力を最適化できます。
[Multiple SSG mode]	デフォルトでは、すべてのオペレーターは単一のスロット共有グループ (SSG) に属しているため、個別のリソース設定ができません。特定のオペレーターのリソースを設定するには、Multiple SSG mode を有効にする必要があります。これにより、各オペレーターに独自の独立したスロットが割り当てられ、そのリソースを直接設定できるようになります。

(オプション) スクリプトパラメーターの設定

Real-Time configuration パネルの Script Parameters セクションで、Add parameters をクリックし、Parameter name と Parameter Value を入力します。これにより、コードでパラメーターを動的に使用できます。

(オプション) Flink 実行パラメーターの設定

Real-Time configuration パネルの Flink running parameters セクションで、以下のパラメーターを設定します。詳細については、「デプロイ情報を設定する」をご参照ください。

パラメーター	説明
[System Checkpoint Interval]	Flink デプロイがチェックポイントを作成する間隔。間隔を短くすると障害後の復旧時間が短縮されますが、システムオーバーヘッドが増加します。このパラメーターを空のままにすると、チェックポイント機能が無効になります。
[Minimum time interval between two system checkpoints]	チェックポイント間に経過する必要がある最小時間。この設定により、頻繁なチェックポイントがシステムパフォーマンスに影響を与えるのを防ぎます。特にチェックポイントの並列度が 1 の場合、1 つのチェックポイントの完了から次のチェックポイントの開始までの最小遅延を保証します。
[状態 TTL]	Flink デプロイ内の状態データの Time-to-Live (TTL)。アクセスまたは更新されずに状態を保持できる最大期間を決定します。デフォルト値は 36 時間で、この時間が経過すると状態は期限切れとなり削除されます。これにより、ストレージとリソース使用率を最適化できます。重要デフォルト値はクラウドのベストプラクティスに基づいており、状態情報が期限切れにならないことを意味するオープンソースのデフォルト値 0 とは異なります。
[その他の設定]	その他の Flink 実行パラメーターを設定できます。例：`taskmanager.network.memory.max:4g` 説明パラメーター設定の詳細については、「デプロイ情報の設定」をご参照ください。

ノードを設定した後、Saveをクリックします。

ステップ 3：(オプション) Flink SQL ストリーミングノードのデバッグ

ノードを本番環境に公開する前に、デバッグ機能を使用して、アップロードしたモックデータでコードをテストできます。これにより、タスクをオペレーションセンターに公開することなく、SQL ロジックとデータフローを検証できます。

説明

デバッグ機能は、許可リストに登録されたユーザーが利用できます。この機能を使用するには、チケットを送信して有効化を依頼してください。

Flink リソースの設定

Run Configuration パネルの Flink resource information セクションで、以下のパラメーターを設定します。

パラメーター	説明
[Flink Debug Cluster]	デバッグタスクの実行に使用する Flink セッションクラスターです。このパラメーターは必須です。ドロップダウンリストには、現在のコンピューティングリソース配下にある既存のセッションクラスターとそのステータスが表示されます。[実行中] 状態のクラスターのみ選択できます。利用可能なクラスターがない場合は、Create Cluster をクリックして Realtime Compute for Apache Flink コンソールに移動し、新しいセッションクラスターを作成します。
[Flink Engine Version]	選択したセッションクラスターの Flink エンジンバージョン。この値はシステムによって自動的に入力され、手動で編集することはできません。
[Timeout]	1 回のデバッグタスクの最大実行時間 (分単位)。デフォルトは 30 分です。この制限を超えると、タスクは自動的に停止されます。

説明

現在のノードの計算リソースを切り替えると、選択されている Flink Debug Cluster およびアップロード済みのデバッグデータがクリアされます。新しいクラスターを選択し、データを再アップロードする必要があります。

デバッグデータの準備

Run Configuration パネルの Debug Data セクションで、コードで参照されているソーステーブルのモックデータを準備します。

Generate Template をクリックします。システムは現在の SQL で参照されているソーステーブルを解析し、一覧表示します。以前にアップロードされたデータはクリアされません。
テーブルの Actions 列で Download Template をクリックすると、ソーステーブルのスキーマと一致する CSV テンプレートをダウンロードできます。
ローカルマシンで、フィールドの順序に従ってテンプレートにデバッグデータを入力し、CSV ファイルとして保存します。
テーブルの Actions 列で、Upload をクリックし、準備した CSV ファイルを選択します。アップロードが成功すると、Status が Enabled に変わります。
(任意) アップロード後、Preview をクリックすると、下部パネルでデータを表示できます。データを変更するには、CSV ファイルを再アップロードして既存のデータを上書きします。
ソーステーブルのモックデータをデバッグ実行から一時的に除外するには、Disable をクリックします。ステータスが Disabled に変わります。再度含めるには、Enable をクリックします。ステータスが Enabled のデータのみがデバッグ実行で使用されます。

説明

モックデータをアップロードする前に、Flink Debug Cluster を選択する必要があります。選択しない場合、最初にコンピューティングリソースを選択してくださいというプロンプトが表示されます。

デバッグデータは CSV 形式である必要があり、各ファイルは 1 MB を超えることはできません。CSV ファイルの最初の行にはフィールド名を含める必要があります。UTF-8 エンコーディングを推奨します。

デバッグタスクの実行

デバッグデータを準備したら、エディターのツールバーにある Run アイコンをクリックするか、F8 キーを押します。システムは、コード、モックデータ、および Flink リソース情報を、選択したセッションクラスターに送信して実行します。

コードで${variable_name}形式のパラメーターを使用する場合、Script Parametersセクションで値を割り当てる必要があります。デバッグ時に、システムはコードを送信する前にこれらのプレースホルダーを指定された値に置き換えます。

デバッグ結果の表示

デバッグタスクの実行後、ページ下部の結果パネルに次の情報が表示され、問題を迅速に特定できます。

コード：今回の実行で Flink エンジンに送信された SQL コード。すべての変数がその値に置き換えられています。
ログ：デバッグタスクの実行ログとエラーメッセージ。
クエリ結果：デバッグタスクの出力データ。

ステップ 4：Flink SQL ストリーミングノードの開始

Flink SQL ストリーミングノードを公開します。
タスクを実行する前に、オペレーションセンターに公開する必要があります。画面の指示に従って、Flink SQL ストリーミングノードを公開します。詳細については、「ノードとワークフローの公開」をご参照ください。
説明
この操作により、タスクは Ververica Platform (VVP) にも公開されます。DataWorks から公開されたタスクは、VVP コンソールの [デプロイ] ページで確認できます。
Flink SQL ストリーミングノードを開始します。
ノードを公開した後、[本番環境に公開] の下にある Go to operation and maintenance をクリックします。ノード O&M で、Real-time Task O&M > Real-time computing tasks > Operation に移動します。タスクを見つけ、[アクション] 列の Start をクリックしてタスクを実行し、そのステータスを監視します。