すべてのプロダクト
Search

このプロダクト

    Realtime Compute for Apache Flink:PyFlink ジョブ

    ドキュメントセンター

    Realtime Compute for Apache Flink:PyFlink ジョブ

    最終更新日:Jun 22, 2026

    このトピックでは、Realtime Compute for Apache Flink でのストリーミングおよびバッチ PyFlink ジョブのデプロイと開始方法について説明し、開発ワークフローを解説します。

    前提条件

    • Resource Access Management (RAM) ユーザーまたは RAM ロールを使用してコンソールにアクセスする場合、その ID に必要な権限が付与されていることを確認してください。詳細については、「権限管理」をご参照ください。

    • ワークスペースが作成されていること。詳細については、「Realtime Compute for Apache Flink の有効化」をご参照ください。

    ステップ 1： Python コードファイルの準備

    Realtime Compute for Apache Flink の管理コンソールでは、Python 開発環境は提供されていません。ジョブはローカルで開発してください。ジョブのデバッグとコネクタの詳細については、「PyFlink ジョブの開発」をご参照ください。

    重要

    ローカル開発で使用する Flink のバージョンが、「ステップ 3： PyFlink ジョブのデプロイ」で選択するエンジンバージョンと一致していることを確認してください。カスタム Python 仮想環境、サードパーティの Python パッケージ、JAR パッケージ、データファイルなどの他の依存関係の使用方法については、「Python の依存関係の使用」をご参照ください。

    すぐに開始できるよう、このトピックではワードカウントジョブのサンプル Python ファイルとサンプルデータファイルを提供しています。これらをダウンロードして、次のステップで使用できます。

    • Shakespeare をクリックして、サンプルデータファイルをダウンロードします。

    ステップ 2： Python ファイルとデータファイルのアップロード

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

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

    3. 左側のナビゲーションウィンドウで、アーティファクト をクリックします。

    4. アーティファクトのアップロード をクリックして、Python ファイルとデータファイルをアップロードします。

      ステップ 1 でダウンロードしたサンプルの Python ファイルとデータファイルをアップロードしてください。ファイルストレージパスの詳細については、「アーティファクト」をご参照ください。

    ステップ 3： PyFlink ジョブのデプロイ

    ストリーミング

    1. O&M > デプロイメント ページで、デプロイメントの作成 > Python デプロイメント を選択します。

    2. デプロイメントパラメーターを設定します。

      パラメーター

      説明

      例

      デプロイメントモード

      ストリームモードを選択します。

      ストリームモード

      デプロイメント名

      Python デプロイメントの名前を入力します。

      flink-streaming-test-python

      エンジンバージョン

      デプロイメントの Flink エンジンバージョン。

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

      vvr-8.0.9-flink-1.17

      Python URI

      word_count_streaming.py サンプルファイルをダウンロードします。次に、アップロードアイコン 上传 をクリックしてファイルを選択し、アップロードします。

      ファイルがすでに アーティファクト に存在する場合、再アップロードせずに直接選択できます。

      -

      エントリーモジュール

      プログラムのエントリーポイントモジュール。

      • PyFlink ジョブが .py ファイルの場合、このパラメーターは必須ではありません。

      • PyFlink ジョブが .zip ファイルの場合、エントリーモジュールを入力する必要があります。例：word_count。

      必須ではありません

      エントリーポイントの main 引数

      main メソッドに渡す引数。

      このチュートリアルでは、入力データファイル Shakespeare のストレージパスを入力します。

      --input oss://<Your-OSS-Bucket-Name>/artifacts/namespaces/<Your-Workspace-Name>/Shakespeare

      [アーティファクト] ページから Shakespeare ファイルの完全なパスをコピーできます。

      デプロイメントターゲット

      ドロップダウンリストから、ターゲットの キュー または セッションクラスター を選択します。セッションクラスターは本番環境での使用を推奨しません。詳細については、「キューの管理」および「セッションクラスターの作成」をご参照ください。

      重要

      セッションクラスター上のデプロイメントは、監視メトリクス、アラート設定、または Autopilot をサポートしていません。セッションクラスターは開発およびテスト目的でのみ使用し、本番環境では使用しないでください。詳細については、「デプロイメントのデバッグ」をご参照ください。

      default-queue

      その他の構成パラメーターの詳細については、「ジョブのデプロイ」をご参照ください。

    3. デプロイ をクリックします。

    バッチ

    1. O&M > デプロイメント ページで、デプロイメントの作成 をクリックし、Python デプロイメント を選択します。

    2. デプロイメントパラメーターを設定します。

      パラメーター

      説明

      例

      デプロイメントモード

      バッチモードを選択します。

      バッチモード

      デプロイメント名

      Python デプロイメントの名前を入力します。

      flink-batch-test-python

      エンジンバージョン

      デプロイメントの Flink エンジンバージョン。

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

      vvr-8.0.9-flink-1.17

      Python URI

      word_count_batch.py サンプルファイルをダウンロードします。次に、アップロードアイコン 上传 をクリックしてファイルを選択し、アップロードします。

      -

      エントリーモジュール

      プログラムのエントリーポイントモジュール。

      • PyFlink ジョブが .py ファイルの場合、このパラメーターは必須ではありません。

      • PyFlink ジョブが .zip ファイルの場合、エントリーモジュールを入力する必要があります。例：word_count。

      必須ではありません

      エントリーポイントの main 引数

      main メソッドに渡す引数。

      このチュートリアルでは、入力ファイル Shakespeare と出力ディレクトリ python-batch-quickstart-test-output のストレージパスを入力します。

      説明

      出力ディレクトリのパスを指定するだけで済みます。出力ディレクトリは、入力ファイルと同じ親ディレクトリにある必要があります。出力ディレクトリを事前に作成する必要はありません。

      --input oss://<Your-OSS-Bucket-Name>/artifacts/namespaces/<Your-Workspace-Name>/Shakespeare

      --output oss://<Your-OSS-Bucket-Name>/artifacts/namespaces/<Your-Workspace-Name>/python-batch-quickstart-test-output

      Shakespeare ファイルの完全なパスは、アーティファクト ページからコピーできます。

      デプロイメントターゲット

      ドロップダウンリストから、ターゲットの キュー または セッションクラスター を選択します。セッションクラスターは本番環境での使用を推奨しません。詳細については、「キューの管理」および「セッションクラスターの作成」をご参照ください。

      重要

      セッションクラスター上のデプロイメントは、監視メトリクス、アラート設定、または Autopilot をサポートしていません。セッションクラスターは開発およびテスト目的でのみ使用し、本番環境では使用しないでください。詳細については、「デプロイメントのデバッグ」をご参照ください。

      default-queue

      その他の構成パラメーターの詳細については、「ジョブのデプロイ」をご参照ください。

    3. デプロイ をクリックします。

    ステップ 4： デプロイメントの開始と結果の表示

    ストリーミング

    1. O&M > デプロイメント ページで、対象のデプロイメントを見つけ、操作 列の 開始 をクリックします。

    2. ジョブの開始 ダイアログボックスで、初期モード を選択し、開始 をクリックします。詳細については、「デプロイメントの開始」をご参照ください。

      実行中 または 完了 のステータスは、デプロイメントが期待どおりに実行されていることを示します。このトピックのサンプルファイルを使用する場合、最終的なステータスは 完了 になります。

    3. デプロイメントステータスが [RUNNING] に変更されたら、ストリーミングデプロイメントの結果を表示します。

      重要

      このトピックのサンプル Python ファイルを使用する場合、ストリーミングデプロイメントが 完了 状態になると結果が削除されるため、結果はデプロイメントが 実行中 状態のときにのみ表示できます。

      .out で終わる TaskManager ログファイルで、shakespeare を検索して計算結果を見つけます。

      ログ タブで、実行中の TaskManager タブをクリックします。関連する TaskManager の ログリスト サブタブをクリックします。flink.out ファイルを開き、右上の検索ボックスに shakespeare と入力して、(shakespeare,1) のようなワードカウント結果を見つけます。

    バッチ

    1. O&M > デプロイメント ページで、対象のデプロイメントを見つけ、操作列の 開始 をクリックします。

      リストをフィルターするには、タイプのドロップダウンリストから バッチデプロイメント を選択します。

    2. ジョブの開始 ダイアログボックスで、開始 をクリックします。詳細については、「デプロイメントの開始」をご参照ください。

    3. デプロイメントのステータスが 完了 に変わったら、バッチデプロイメントの結果を表示します。

      OSS コンソールにログインします。oss://<Your-OSS-Bucket-Name>/artifacts/namespaces/<Your-Workspace-Name>/python-batch-quickstart-test-output ディレクトリに移動します。デプロイメントの開始日時で名付けられたフォルダーをクリックし、対象のファイルをクリックしてから、表示されるパネルで ダウンロード をクリックします。

      下载

      バッチデプロイメントは .ext ファイルを生成します。ファイルをダウンロードした後、テキストエディターまたは Microsoft Word で開いて結果を表示します。出力は次のようになります：

      (As,40)
(At,5)
(Ay,1)
(Be,9)
(By,14)
(Do,4)
(He,7)
(I,,4)
(If,34)
(In,36)
(Is,10)
(It,6)

    (任意) ステップ 5： デプロイメントの停止

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

    関連トピック