Flink SQL バッチノードを使用すると、標準 SQL ステートメントを使用してデータ処理タスクを定義および実行できます。 データクレンジングや集計などのタスクにおいて、大規模なデータセットを分析および変換するために使用します。 このノードは、ビジュアル設定をサポートし、大規模なバッチ処理のための効率的で柔軟なソリューションを提供します。 このトピックでは、Flink SQL バッチノードを使用してデータをバッチ処理する方法について説明します。
前提条件
[管理] でワークスペースを作成し、Realtime Compute for Apache Flink コンピューティングリソースをバインド済みであること。 詳細については、「コンピューティングリソースのバインド」をご参照ください。
Flink SQL バッチノードを作成済みであること。 詳細については、「スケジューリングワークフローのノード作成」をご参照ください。
DataWorks が Realtime Compute for Apache Flink API を呼び出すために使用する RAM ユーザーまたは RAM ロールに、次の API 権限が付与済みであること。 これらの権限は、ノードタスクを 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} 形式を使用して変数を定義できます。 次に、ノード編集ページの右側にあるScheduling Settings ペインの Scheduling Parameters セクションで、変数に値を割り当てます。 これにより、スケジューリングシナリオでコードにパラメーターを動的に渡すことができます。 スケジューリングパラメーターの使用方法の詳細については、「スケジューリングパラメーターのソースとその式」をご参照ください。 以下に例を示します。
-- datagen_source_${var} という名前のソーステーブルを作成します。
CREATE TEMPORARY TABLE datagen_source_${var}(
name VARCHAR
) WITH (
'connector' = 'datagen',
'number-of-rows' = '1000'
);
-- blackhole_sink_${var} という名前の結果テーブルを作成します。
CREATE TEMPORARY TABLE blackhole_sink_${var}(
name VARCHAR
) WITH (
'connector' = 'blackhole'
);
-- ソーステーブルのデータを結果テーブルに挿入します。
INSERT INTO blackhole_sink_${var}
SELECT
name
FROM datagen_source_${var};この例では、パラメーター bizdate の値が $[yyyymmdd] であるため、新しい日次データのバッチ同期が可能になります。
ステップ 2：Flink SQL バッチノードの設定
ビジネス要件に基づいて、Flink SQL バッチノードのタスクパラメーターを設定します。
Flink リソースの設定
編集ページの右側、Scheduling Settings の下の Flink resource information セクションで、以下のパラメーターを設定できます。 詳細については、「ジョブデプロイ情報を設定する」をご参照ください。
パラメーター | 説明 |
[Flink cluster] | [管理] でバインドしたフルマネージドの Flink コンピューティングリソースの名前。 |
[Flink engine version] | 要件に基づいてエンジンバージョンを選択します。 |
[Resource Group for Scheduling] | Flink クラスターとネットワーク接続が可能なサーバーレスリソースグループを選択します。 |
JobManager CPU | 安定した運用のために、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 ジョブにおける並列タスク実行数を指定します。 同時実行数を高くすると、処理速度とリソース使用率が向上します。 クラスターリソースとジョブの特性に基づいて、適切な値を設定してください。 |
[Maximum number of slots] | スロットは、TaskManager 上の固定サイズのリソースセットであり、タスクに割り当てることができます。 各スロットは、1 つのタスクまたは演算子インスタンスを実行できます。 利用可能なリソースに基づいて、最大スロット数を調整できます。 |
[Number of slots per TaskManager] | TaskManager あたりのスロット数を指定します。これにより、並列実行できるタスクの数が決まります。 この設定を調整することで、リソース使用率と並列処理を最適化できます。 |
(オプション) スケジューリングパラメーターの設定
編集ページの右側にあるScheduling SettingsのScheduling Parametersセクションで、Add parametersをクリックし、Parameter nameとParameter Valueを編集することで、コード内で動的に使用できます。
(オプション) Flink 実行時パラメーターの設定
編集ページの右側にあるScheduling SettingsセクションのFlink running parametersエリアで、ランタイムパラメーターを設定できます。詳細については、「ジョブデプロイメント情報の設定」をご参照ください。
Flink 実行時パラメーターは VVP (Ververica Platform) と互換性があり、YAML 形式で指定します。 改行のためにセミコロンなどの特殊文字を追加する必要はありません。
ノードタスクを定期的に実行するには、ビジネス要件に基づいて、スケジューリング情報 (Scheduling Policy、Scheduling time、Scheduling Dependency、および Node output parameters) を設定します。 詳細については、「ノードスケジューリング設定」をご参照ください。
タスク設定を完了したら、Save をクリックします。
ステップ 3：(オプション) Flink SQL バッチノードのデバッグ
ノードを本番環境にデプロイする前に、デバッグ機能を使用して、アップロードされたモックデータでノードコードをテストできます。 これにより、タスクをオペレーションセンターにデプロイすることなく、事前に SQL ロジックを検証できます。
デバッグ機能は、許可リストに登録されたユーザーのみが利用できます。 この機能を使用するには、チケットを送信してアクセスをリクエストしてください。
Flink リソース情報の設定
ノード編集ページの右側にある Run Configuration パネルの Flink resource information セクションで、次の表に従ってパラメーターを設定します。
パラメーター | 説明 |
[Flink Debug Cluster] | デバッグタスクの実行に使用される Flink セッションクラスターです。このパラメーターは必須です。ドロップダウンリストには、現在のコンピューティングリソース配下にある既存のセッションクラスターとそのステータスが表示されます。ステータスが [実行中] のクラスターのみ選択できます。 リストに利用可能なクラスターがない場合は、Create Cluster をクリックしてリアルタイムコンピューティング for Apache Flink コンソールに移動し、新しいセッションクラスターを作成できます。 |
[Flink Engine Version] | 選択したセッションクラスターの Flink エンジンバージョンです。 この値はシステムによって自動的に入力されるため、設定する必要はありません。 |
[Timeout] | 単一のデバッグタスクの最大実行時間 (分単位)。 デフォルト値は 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 のリソース情報を選択した Session Cluster に送信して実行します。
コードで ${variable_name} 形式のパラメーターを使用する場合は、Scheduling Parameters セクションで変数に値を割り当ててください。デバッグ中、システムは提出前にコード内の同じ名前のプレースホルダーをこの値に置き換えます。
デバッグ結果の表示
デバッグタスクの実行後、ノードページの下部にある結果パネルに、問題のトラブルシューティングに役立つ次の情報が表示されます：
コード： Flink エンジンに送信された SQL コード (変数置換後)。
ログ： デバッグタスクの実行時ログと例外。
クエリ結果： デバッグタスクの出力データ。