ジョブを開発した後、デプロイする必要があります。デプロイメントによって、開発環境と本番環境が分離されます。デプロイメントは実行中のジョブに影響を与えず、ジョブが開始または再起動された後にのみオンラインになります。このトピックでは、SQL、YAML、JAR、Python ジョブのデプロイ方法について説明します。
前提条件
ジョブが開発済みであること。
SQL ジョブについては、「ジョブ開発マップ」をご参照ください。
YAML ジョブについては、「Flink CDC データインジェストジョブの開発 (パブリックプレビュー)」をご参照ください。
Python ジョブの場合、Python パッケージを開発する必要があります。詳細については、「Python ジョブの開発」をご参照ください。
JAR ジョブの場合、JAR パッケージを開発する必要があります。詳細については、「JAR ジョブの開発」をご参照ください。
リソースのアップロード
ジョブをデプロイする前に、必要な JAR パッケージ、Python ジョブファイル、または Python 依存関係を Flink 開発コンソールにアップロードする必要があります。
対象のワークスペースを見つけ、[操作] 列の [コンソール] をクリックします。
左側のナビゲーションウィンドウで、[ファイル管理] をクリックします。
[リソースのアップロード] をクリックし、アップロードする JAR パッケージ、Python ジョブファイル、または Python 依存関係を選択します。
ジョブが Python API ジョブの場合は、PyFlink の公式 JAR パッケージをアップロードします。公式 JAR パッケージのダウンロード URL の詳細については、「PyFlink V1.11」および「PyFlink V1.12」をご参照ください。
操作手順
対象のワークスペースを見つけ、[操作] 列の [コンソール] をクリックします。ジョブをデプロイする手順は、ジョブの種類によって異なります。
JAR ジョブのデプロイ
ページで、 を選択します。
次の設定情報を入力します。
パラメーター
説明
デプロイメントモード
ストリームまたはバッチを選択します。
デプロイメント名
JAR ジョブの名前を入力します。
エンジンバージョン
エンジンバージョンの詳細については、「エンジンバージョン」および「ライフサイクルポリシー」をご参照ください。推奨または安定バージョンを使用することを推奨します。バージョンタグは次のように説明されます：
推奨：最新のメジャーバージョンの最新のマイナーバージョン。
安定：サービス期間内にあるメジャーバージョンの最新のマイナーバージョン。以前のバージョンのバグはこのバージョンで修正されています。
通常：サービス期間内にあるその他のマイナーバージョン。
EOS：サービス終了 (EOS) に達したバージョン。
[JAR URI]
ファイルを選択するか、新しいファイルをアップロードします。ファイルをこのエリアにドラッグするか、
アイコンをクリックしてアップロードするファイルを選択できます。説明Realtime Compute for Apache Flink VVR 8.0.6 以降は、Flink ワークスペースにバインドされたバケットへのアクセスのみをサポートします。他のバケットへのアクセスはサポートされていません。
ジョブが Python API ジョブの場合は、PyFlink の公式 JAR パッケージを指定します。公式 JAR パッケージのダウンロード URL の詳細については、「PyFlink V1.11」および「PyFlink V1.12」をご参照ください。
エントリーポイントクラス
プログラムのエントリクラス。JAR パッケージにメインクラスが指定されていない場合は、[エントリポイントクラス] の標準パスを入力します。
説明ジョブが Python API ジョブの場合、エントリポイントクラスを `org.apache.flink.client.python.PythonDriver` に設定します。
エントリポイントのメイン引数
ここでパラメーターを渡し、main メソッドで呼び出すことができます。
説明パラメーター情報は 1024 文字を超えることはできません。改行、スペース、その他の特殊文字を含むような複雑なパラメーターは渡さないでください。複雑なパラメーターを渡すには、追加の依存ファイルを使用します。
ジョブが Python API ジョブの場合は、まず Python ジョブファイルをアップロードする必要があります。Python ジョブファイルがアップロードされると、デフォルトでジョブが実行されるノードの `/flink/usrlib/` ディレクトリに保存されます。
たとえば、Python ジョブファイルの名前が `word_count.py` の場合、エントリポイントのメイン引数を
-py /flink/usrlib/word_count.pyに設定します。Python ジョブファイルの完全なパスを指定する必要があります。`/flink/usrlib/` ディレクトリは省略したり変更したりすることはできません。
追加の依存関係ファイル
(推奨) アップロード済みの追加の依存ファイルを選択します。
事前に依存関係ファイルをアップロードする必要があります。ファイルをアップロードするには、Realtime Compute for Apache Flink の開発コンソールの左側にあるナビゲーションウィンドウで、[Resource Management] に移動します。または、デプロイメントを作成するときに、[追加の依存関係] の右側にある
アイコンをクリックします。アップロードした依存関係ファイルは、ワークスペースのアクティブ化時に Realtime Compute for Apache Flink ワークスペースに関連付けた Object Storage Service (OSS) バケットの artifacts ディレクトリに保存されます。ファイルパスは oss://<Name of your associated OSS bucket>/artifacts/namespaces/<namespace name>のフォーマットです。ターゲットの追加依存ファイルの OSS パスを入力します。
OSS パスは、現在の Flink ワークスペースを作成したときに選択した OSS バケットを指している必要があります。
ターゲットの追加依存ファイルの URL を入力します。
URL は、Realtime Compute for Apache Flink がアクセスできる外部ストレージシステムのアドレスである必要があります。アクセスは許可されている必要があります (例：公開読み取り、または権限不要)。ファイル名で終わる URL のみがサポートされます (例：http://xxxxxx/<file>)。
説明3 つの方法のいずれかを使用してアップロードされた追加の依存ファイルも、宛先マシンにダウンロードされます。ジョブの実行時に、ファイルは JobManager (JM) と TaskManager (TM) が存在する Pod の /flink/usrlib/ ディレクトリにロードされます。
[デプロイメントターゲット] としてセッションクラスターを選択した場合、ジョブに追加の依存ファイルを設定することはできません。
デプロイメントターゲット
ドロップダウンリストから、ターゲットの [リソースキュー] または [セッションクラスター] (本番環境では使用不可) を選択します。詳細については、「リソースキューの管理」および「ステップ 1: セッションクラスターの作成」をご参照ください。
説明セッションクラスターにデプロイされたジョブは、モニタリングとアラート、または自動チューニングをサポートしていません。本番環境ではセッションクラスターを使用しないでください。セッションクラスターはステージング環境として使用できます。詳細については、「ジョブのデバッグ」をご参照ください。
説明
オプション。ジョブの説明を入力します。
[ジョブタグ]
ジョブにタグを設定すると、[ジョブ O&M] ページで [タグキー] と [タグ値] によってジョブをすばやく見つけることができます。ジョブには最大 3 つのタグを作成できます。
その他の設定
このスイッチをオンにして、次のパラメーターを設定します：
[Kerberos クラスター]：ドロップダウンリストから、作成した Kerberos クラスターを選択します。Kerberos クラスターの作成方法の詳細については、「Kerberos クラスターの作成」をご参照ください。
[principal]：Kerberos プリンシパルはユーザーまたはサービスです。Kerberos 暗号化システムで ID を一意に識別するために使用されます。
[デプロイ] をクリックします。
[ジョブ O&M] ページで、デプロイされた JAR ジョブを表示し、必要に応じて開始できます。
Python ジョブのデプロイ
ページで、 を選択します。
パラメーターを設定します。
パラメーター
説明
デプロイメントモード
ストリームモードまたはバッチモードでデプロイできます。
デプロイメント名
Python ジョブの名前を入力します。
エンジンバージョン
エンジンバージョンの詳細については、「エンジンバージョン」および「ライフサイクルポリシー」をご参照ください。推奨または安定バージョンを使用することを推奨します。バージョンタグは次のように説明されます：
推奨：最新のメジャーバージョンの最新のマイナーバージョン。
安定：サービス期間内にあるメジャーバージョンの最新のマイナーバージョン。以前のバージョンのバグはこのバージョンで修正されています。
通常：サービス期間内にあるその他のマイナーバージョン。
EOS：サービス終了 (EOS) に達したバージョン。
Python ファイルパス
Python ジョブファイルを選択します。
Python ファイルは .py または .zip ファイルです。
[エントリモジュール] パラメーターが空の場合、このパラメーターは .py ファイルである必要があります。
エントリーモジュール
プログラムのエントリクラス (例：example.word_count)。
[Python ファイルパス] が .zip ファイルの場合、このパラメーターは必須です。
エントリーポイントのメイン引数
ジョブのパラメーター。
Python ライブラリ
サードパーティの Python パッケージ。パッケージは Python ワーカープロセスの PYTHONPATH に追加され、Python ユーザー定義関数 (UDF) で直接アクセスできます。サードパーティの Python パッケージの使用方法の詳細については、「サードパーティの Python パッケージの使用」をご参照ください。
Python アーカイブ
アーカイブファイル。ZIP 形式のファイルのみがサポートされます (例：.zip、.jar、.whl、.egg ファイル)。
アーカイブファイルは、Python ワーカープロセスの作業ディレクトリに解凍されます。たとえば、アーカイブファイルが mydata.zip という名前の圧縮パッケージにある場合、Python UDF で次のコードを記述して mydata.zip アーカイブファイルにアクセスできます。
def map(): with open("mydata.zip/mydata/data.txt") as f: ...Python アーカイブの詳細については、「カスタム Python 仮想環境の使用」および「データファイルの使用」をご参照ください。
追加の依存関係ファイル
Python ジョブファイルと依存データファイルを選択します。
(推奨) アップロード済みの追加の依存ファイルを選択します。
事前に依存関係ファイルをアップロードする必要があります。ファイルをアップロードするには、Realtime Compute for Apache Flink の開発コンソールの左側にあるナビゲーションウィンドウで [Resource Management] に移動します。または、デプロイメントを作成するときに、[追加の依存関係] の右側にある
アイコンをクリックします。アップロードした依存関係ファイルは、ワークスペースを有効化する際に Realtime Compute for Apache Flink ワークスペースに関連付けるために選択した Object Storage Service (OSS) バケットのアーティファクトディレクトリに保存されます。ファイルパスは oss://<Name of your associated OSS bucket>/artifacts/namespaces/<namespace name>のフォーマットです。ターゲットの追加依存ファイルの OSS パスを入力します。
OSS パスは、現在の Flink ワークスペースを作成したときに選択した OSS バケットを指している必要があります。
ターゲットの追加依存ファイルの URL を入力します。
URL は、Realtime Compute for Apache Flink がアクセスできる外部ストレージシステムのアドレスである必要があります。アクセスは許可されている必要があります (例：公開読み取り、または権限不要)。ファイル名で終わる URL のみがサポートされます (例：http://xxxxxx/<file>)。
説明3 つの方法のいずれかを使用してアップロードされた追加の依存ファイルも、宛先マシンにダウンロードされます。ジョブの実行時に、ファイルは JobManager (JM) と TaskManager (TM) が存在する Pod の /flink/usrlib/ ディレクトリにロードされます。
[デプロイメントターゲット] としてセッションクラスターを選択した場合、ジョブに追加の依存ファイルを設定することはできません。
重要JAR パッケージの依存関係を使用する場合は、デプロイ後に
pipeline.classpathsパラメーターを設定して、JAR パッケージの依存関係を参照する必要があります。詳細については、「JAR パッケージの依存関係の使用」をご参照ください。デプロイメントターゲット
ドロップダウンリストから、ターゲットの [リソースキュー] または [セッションクラスター] (本番環境では使用不可) を選択します。詳細については、「リソースキューの管理」および「ステップ 1: セッションクラスターの作成」をご参照ください。
説明セッションクラスターにデプロイされたジョブは、モニタリングとアラート、または自動チューニングをサポートしていません。本番環境ではセッションクラスターを使用しないでください。セッションクラスターはステージング環境として使用できます。詳細については、「ジョブのデバッグ」をご参照ください。
注
オプションで説明を入力できます。
[ジョブタグ]
ジョブにタグを設定すると、[ジョブ O&M] ページで [タグキー] と [タグ値] によってジョブをすばやく見つけることができます。ジョブには最大 3 つのタグを作成できます。
その他の設定
このスイッチをオンにして、次のパラメーターを設定します：
[Kerberos クラスター]：ドロップダウンリストから、作成した Kerberos クラスターを選択します。Kerberos クラスターの作成方法の詳細については、「Kerberos クラスターの作成」をご参照ください。
[principal]：Kerberos プリンシパルはユーザーまたはサービスです。Kerberos 暗号化システムで ID を一意に識別するために使用されます。
[デプロイ] をクリックします。
[ジョブ O&M] ページで、デプロイされた Python ジョブを表示し、必要に応じて開始できます。
関連ドキュメント
ジョブがオンラインになる前または後で、ジョブのリソースとデプロイメント情報を設定できます。詳細については、「ジョブのデプロイメント情報の設定」および「ジョブリソースの設定」をご参照ください。
ジョブをデプロイした後、 ページでジョブを開始してオンラインで実行する必要があります。ジョブの開始方法の詳細については、「ジョブの開始」をご参照ください。