すべてのプロダクト
Search

このプロダクト

    Realtime Compute for Apache Flink:Flink SQL ジョブ

    ドキュメントセンター

    Realtime Compute for Apache Flink:Flink SQL ジョブ

    最終更新日:Jun 23, 2026

    このトピックでは、Flink SQL ジョブの作成、デプロイ、および開始の方法を説明し、基本的な開発および運用ワークフローの概要を説明します。

    前提条件

    • ご利用の RAM ユーザーまたは RAM ロールには、Realtime Compute コンソールに必要な権限が付与されていること。詳細については、「権限」をご参照ください。

    • Flink ワークスペースが作成されていること。詳細については、「Realtime Compute for Apache Flink のアクティベーション」をご参照ください。

    ステップ 1：SQL ドラフトの作成

    1. SQL ドラフト作成ページに移動します。

      1. Realtime Compute コンソールにログインします。

      2. 目的の Flink ワークスペースを見つけ、[操作] 列の [コンソール] をクリックします。

      3. ナビゲーションウィンドウで、開発 > ETL をクリックします。

    2. image アイコンをクリックし、次に [新規ブランクストリームドラフト] をクリックします。[ファイル名] を入力し、[エンジンバージョン] を選択します。

      Realtime Compute for Apache Flink は、さまざまなコードテンプレートとデータ同期テンプレートを提供しています。各テンプレートには、ユースケースの説明、コード例、および手順が含まれています。テンプレートをクリックすると、プロダクトの機能と構文をすばやく学習し、ビジネスロジックを実装できます。詳細については、「コードテンプレート」および「データ同期テンプレート」をご参照ください。

      パラメーター

      説明

      例

      ファイル名

      SQL ドラフトの名前。

      説明

      名前は現在のプロジェクト内で一意である必要があります。

      flink-test

      エンジンバージョン

      SQL ドラフトの Flink エンジンバージョン。

      信頼性とパフォーマンスを向上させるために、推奨 または 安定 タグの付いたバージョンを使用することを推奨します。エンジンバージョンの詳細については、「リリースノート」および「エンジンバージョン」をご参照ください。

      vvr-8.0.8-flink-1.17

    3. 作成 をクリックします。

    ステップ 2：SQL の記述とドラフト構成の表示

    1. SQL コードを記述します。

      次の SQL をエディターにコピーします。この例では、Datagen コネクタを使用してランダムなデータストリームを生成し、Print コネクタを使用して出力をコンソールログに書き込みます。サポートされているコネクタの詳細については、「サポートされているコネクタ」をご参照ください。

      -- datagen_source という名前の一時ソーステーブルを作成します。
CREATE TEMPORARY TABLE datagen_source(
  randstr VARCHAR
) WITH (
  'connector' = 'datagen' -- Datagen コネクタを使用します。
);

-- print_table という名前の一時結果テーブルを作成します。
CREATE TEMPORARY TABLE print_table(
  randstr  VARCHAR
) WITH (
  'connector' = 'print',   -- Print コネクタを使用します。
  'logger' = 'true'        -- 出力をログに書き込みます。
);

-- randstr フィールドから部分文字列を選択し、結果テーブルに挿入します。
INSERT INTO print_table
SELECT SUBSTRING(randstr,0,8) from datagen_source;
      説明

      • この例では、INSERT INTO 文を使用して単一の結果テーブルにデータを書き込みます。INSERT INTO 文を使用して複数の結果テーブルにデータを書き込むこともできます。詳細については、「INSERT INTO 文」をご参照ください。

      • 本番環境では、一時テーブルの代わりにカタログに登録されたテーブルを使用してください。詳細については、「カタログ」をご参照ください。

    2. ドラフト構成を表示します。

      SQL エディターの右側で、いくつかのタブの設定を表示または構成できます。

      タブ

      説明

      その他の設定

      • エンジンバージョン：詳細については、「エンジンバージョン」および「ライフサイクルポリシー」をご参照ください。推奨または安定バージョンを使用することを推奨します。バージョンタグは次のとおりです：

        • 推奨：現在のメジャーバージョンの最新のマイナーバージョン。

        • 安定：サービス期間内であり、既知の不具合が修正されたメジャーバージョンの最新のマイナーバージョン。

        • 通常：サービス期間内にあるその他のマイナーバージョン。

        • 非推奨：サポート終了 (EOL) 日を過ぎたバージョン。

      • 追加の依存関係：一時関数など、ジョブに必要な追加の依存関係。

      • Kerberos 認証：Kerberos 認証を有効にし、登録済みの Kerberos クラスターとプリンシパル情報を構成します。Kerberos クラスターを登録していない場合は、「Hive Kerberos クラスターの登録」をご参照ください。

      コード構造

      • データフロー：データフロー図を使用して、データリネージをすばやく表示します。

      • ツリー構造：ツリー構造図を使用して、データソースをすばやく表示します。

      バージョン情報

      ここで SQL ドラフトのバージョン履歴を表示できます。[アクション] 列の機能の詳細については、「ドラフトバージョンを管理する」をご参照ください。

    (任意) ステップ 3：SQL ドラフトの検証とデバッグ

    1. SQL ドラフトを検証します。

      検証では、SQL のセマンティクス、ネットワーク接続、およびドラフトで使用されるテーブルのメタデータがチェックされます。検証後、結果エリアで SQL 最適化 をクリックすると、潜在的な SQL リスクと最適化の提案を表示できます。

      1. SQL エディターの右上隅にある 詳細チェック をクリックします。

      2. [Deep Check] ダイアログボックスで、[確認] をクリックします。

      説明

      タイムアウトエラーが発生した場合、次のメッセージが表示されることがあります: SQL の解析が複雑すぎるため、RPC がタイムアウトした可能性があります。flink-configuration の flink.sqlserver.rpc.execution.timeout オプション (デフォルトでは 120 s) の値を増やすことを検討してください。

      解決策：SQL エディターの先頭に次の構成パラメーターを追加します。

      SET 'flink.sqlserver.rpc.execution.timeout' = '600s';

    2. SQL ドラフトをデバッグします。

      デバッグ機能を使用すると、ジョブの実行をシミュレートして出力を確認し、SELECT または INSERT ロジックを検証できます。この機能により、開発効率が向上し、データ品質のリスクが軽減されます。

      説明

      デバッグ機能は、結果テーブルにデータを書き込みません。

      1. SQL エディターの右上隅にある デバッグ をクリックします。

      2. [デバッグ] ダイアログボックスで、デバッグクラスターを選択し、[次へ] をクリックします。

        利用可能なセッションクラスターがない場合は、作成する必要があります。セッションクラスターは、SQL ドラフトと同じエンジンバージョンを使用し、実行中である必要があります。詳細については、「ステップ 1：セッションクラスターの作成」をご参照ください。

      3. デバッグデータを構成し、OK をクリックします。

        構成の詳細については、「ステップ 2：ジョブのデバッグ」をご参照ください。

    ステップ 4：SQL ドラフトのデプロイ

    SQL エディターの右上隅で、[デプロイ] をクリックします。[新しいバージョンをデプロイ] ダイアログボックスで、必要に応じてパラメーターを設定し、[OK] をクリックします。

    ドラフトをデプロイする際に、[キュー] または [セッションクラスター] を [デプロイメントターゲット] として選択できます。 次の表では、これら 2 つのオプションを比較します。

    デプロイターゲット

    環境

    主な特徴

    キュー

    本番

    • 排他的リソース：リソースはジョブ専用であり、横取りされないため、安定性が確保されます。

    • リソースの隔離：リソースキューを追加して、リソースを隔離および管理できます。

    • ユースケース：長時間実行されるジョブや優先度の高いジョブに適しています。

    セッションクラスター

    開発およびテスト

    • 共有リソース：複数のジョブが JobManager (JM) を共有し、リソース使用率を向上させます。

    • 高速起動：初期化されたリソースを再利用することで、ジョブが迅速に開始されます。

    • ユースケース：開発、テスト、および軽量なジョブに適しています。リソース共有メカニズムがジョブの安定性に影響を与えないように、リソースクォータを慎重に計画する必要があります。

    重要

    セッションクラスターで実行されるジョブのログは利用できません。

    ステップ 5：ジョブの開始と結果の表示

    1. ナビゲーションウィンドウで、O&M > デプロイメント をクリックします。

    2. 対象のジョブを見つけ、[開始] を [アクション] 列でクリックします。

      ステートレス起動 を選択し、開始 をクリックします。ジョブのステータスが 実行中 に変わると、ジョブは実行されています。開始パラメーターの詳細については、「ジョブの開始」をご参照ください。

    3. [デプロイメント] 詳細ページで、Flink ジョブの結果を表示します。

      1. O&M > デプロイメント ページで、対象のジョブの名前をクリックします。

      2. [ログ] タブで、[実行中の TaskManager] サブタブをクリックします。 [パス、ID] 列で、TaskManager をクリックします。

      3. [ログ] タブをクリックし、PrintSinkOutputWriter に関連するログを検索します。

        チェーン Source: datagen_source → Calc → Sink: print_table を含むログエントリが見つかった場合、データストリームは正しく処理されています。

    (任意) ステップ 6：ジョブの停止

    ジョブに変更 (コードの変更、WITH パラメーターの更新、バージョンの変更など) を適用するには、ジョブを再デプロイし、停止してから再起動する必要があります。ステートレス起動や動的でない構成変更を適用する場合も、再起動が必要です。ジョブの停止の詳細については、「ジョブの停止」をご参照ください。

    1. [O&M] > [デプロイメント] ページで、対象のジョブを見つけ、[操作] 列の [キャンセル] をクリックします。

    2. OK をクリックします。

    関連ドキュメント