Flink アプリケーションを実行する - Amazon EMR

翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

Flink アプリケーションを実行する

Amazon EMR 6.13.0 以降では、EKS 上の Amazon EMR のアプリケーションモードで Flink Kubernetes オペレーターを使用して Flink アプリケーションを実行できます。Amazon EMR 6.15.0 以降では、Flink アプリケーションをセッションモードで実行することもできます。このページでは、EKS 上の Amazon EMR で Flink アプリケーションを実行するために使用できる両方の方法について説明します。

トピック
    注記

    Flink ジョブを送信する際に高可用性メタデータを保存するには、Amazon S3 バケットを作成する必要があります。この機能を使用しない場合には無効にできます。これは、デフォルトでは有効になっています。

    前提条件 - Flink Kubernetes オペレータを使用して Flink アプリケーションを実行する前に、Amazon EMR on EKS での Flink Kubernetes オペレータのセットアップ と Kubernetes オペレータをインストールする のステップを完了してください。

    Application mode

    Amazon EMR 6.13.0 以降では、EKS 上の Amazon EMR のアプリケーションモードで Flink Kubernetes オペレーターを使用して Flink アプリケーションを実行できます。

    1. 以下の例のように FlinkDeployment 定義ファイル basic-example-app-cluster.yaml を作成します。いずれかのオプトイン AWS リージョンを有効にして使用する場合は、必ずコメントを解除して設定してくださいfs.s3a.endpoint.region。

      apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: basic-example-app-cluster
spec:
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "2"
    #fs.s3a.endpoint.region: OPT_IN_AWS_REGION_NAME
    state.checkpoints.dir: CHECKPOINT_S3_STORAGE_PATH
    state.savepoints.dir: SAVEPOINT_S3_STORAGE_PATH 
  flinkVersion: v1_17
  executionRoleArn: JOB_EXECUTION_ROLE_ARN
  emrReleaseLabel: "emr-6.13.0-flink-latest" # 6.13 or higher
  jobManager:
    storageDir: HIGH_AVAILABILITY_STORAGE_PATH
    resource:
      memory: "2048m"
      cpu: 1
  taskManager:
    resource:
      memory: "2048m"
      cpu: 1
  job:
    # if you have your job jar in S3 bucket you can use that path as well
    jarURI: local:///opt/flink/examples/streaming/StateMachineExample.jar
    parallelism: 2
    upgradeMode: savepoint
    savepointTriggerNonce: 0
  monitoringConfiguration:    
    cloudWatchMonitoringConfiguration:
       logGroupName: LOG_GROUP_NAME

    2. 次のコマンドで Flink デプロイを送信します。これにより、basic-example-app-cluster という名前で FlinkDeployment オブジェクトも作成されます。

      kubectl create -f basic-example-app-cluster.yaml -n <NAMESPACE>

    3. Flink UI にアクセスします。

      kubectl port-forward deployments/basic-example-app-cluster 8081 -n NAMESPACE

    4. localhost:8081 を開いて、Flink ジョブをローカルに表示します。

    5. ジョブをクリーンアップします。チェックポイント、高可用性、セーブポイントメタデータ、CloudWatch ログなど、このジョブ用に作成された S3 アーティファクトを忘れずにクリーンアップしてください。

    Flink Kubernetes オペレータを利用して Flink にアプリケーションを送信する方法の詳細については、GitHub の apache/flink-kubernetes-operator フォルダにある「Flink Kubernetes operator examples」を参照してください。

    Session mode

    Amazon EMR 6.15.0 以降では、EKS 上の Amazon EMR のセッションモードで Flink Kubernetes オペレーターを使用して Flink アプリケーションを実行できます。

    1. 以下の例のように basic-example-app-cluster.yaml という名前の FlinkDeployment 定義ファイルを作成します。いずれかのオプトイン AWS リージョンを有効にして使用する場合は、必ずコメントを解除して設定してくださいfs.s3a.endpoint.region。

      apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: basic-example-session-cluster
spec:
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "2"
    #fs.s3a.endpoint.region: OPT_IN_AWS_REGION_NAME
    state.checkpoints.dir: CHECKPOINT_S3_STORAGE_PATH
    state.savepoints.dir: SAVEPOINT_S3_STORAGE_PATH
  flinkVersion: v1_17
  executionRoleArn: JOB_EXECUTION_ROLE_ARN
  emrReleaseLabel: "emr-6.15.0-flink-latest"
  jobManager:
    storageDir: HIGH_AVAILABILITY_S3_STORAGE_PATH
    resource:
      memory: "2048m"
      cpu: 1
  taskManager:
    resource:
      memory: "2048m"
      cpu: 1
  monitoringConfiguration:    
    s3MonitoringConfiguration:
       logUri: 
    cloudWatchMonitoringConfiguration:
       logGroupName: LOG_GROUP_NAME

    2. 次のコマンドで Flink デプロイを送信します。これにより、basic-example-session-cluster という名前で FlinkDeployment オブジェクトも作成されます。

      kubectl create -f basic-example-app-cluster.yaml -n NAMESPACE

    3. 次のコマンドを使用して、セッションクラスター LIFECYCLE が STABLE であることを確認します。

      kubectl get flinkdeployments.flink.apache.org basic-example-session-cluster -n NAMESPACE

      出力は次の例と類似したものになります。

      NAME                              JOB STATUS   LIFECYCLE STATE
basic-example-session-cluster                          STABLE

    4. 次のサンプルコンテンツを含む FlinkSessionJob カスタム定義リソースファイル basic-session-job.yaml を作成します。

      apiVersion: flink.apache.org/v1beta1
kind: FlinkSessionJob
metadata:
  name: basic-session-job
spec:
  deploymentName: basic-session-deployment
  job:
    # If you have your job jar in an S3 bucket you can use that path.
    # To use jar in S3 bucket, set 
    # OPERATOR_EXECUTION_ROLE_ARN (--set emrContainers.operatorExecutionRoleArn=$OPERATOR_EXECUTION_ROLE_ARN)
    # when you install Spark operator
    jarURI: https://repo1.maven.org/maven2/org/apache/flink/flink-examples-streaming_2.12/1.16.1/flink-examples-streaming_2.12-1.16.1-TopSpeedWindowing.jar
    parallelism: 2
    upgradeMode: stateless

    5. 次のコマンドで Flink セッションジョブを送信します。これにより、FlinkSessionJob オブジェクト basic-session-job が作成されます。

      kubectl apply -f basic-session-job.yaml -n $NAMESPACE

    6. 次のコマンドを使用して、セッションクラスター LIFECYCLE が STABLE、JOB STATUS が RUNNING であることを確認します。

      kubectl get flinkdeployments.flink.apache.org basic-example-session-cluster -n NAMESPACE

      出力は次の例と類似したものになります。

      NAME                              JOB STATUS   LIFECYCLE STATE
basic-example-session-cluster     RUNNING      STABLE

    7. Flink UI にアクセスします。

      kubectl port-forward deployments/basic-example-session-cluster 8081 -n NAMESPACE

    8. localhost:8081 を開いて、Flink ジョブをローカルに表示します。

    9. ジョブをクリーンアップします。チェックポイント、高可用性、セーブポイントメタデータ、CloudWatch ログなど、このジョブ用に作成された S3 アーティファクトを忘れずにクリーンアップしてください。