E-MapReduce:Flink ジョブの送信

前提条件

EMR on ACK コンソールで Flink クラスターを作成済みであること。詳細については、「クイックスタート」をご参照ください。

注意事項

このトピックの例では、JAR ファイルはイメージに直接パッケージ化されています。独自の JAR ファイルを使用するには、Object Storage Service (OSS) にアップロードしてください。詳細については、「シンプルアップロード」をご参照ください。

OSS の JAR ファイルを使用する場合は、local:///opt/flink/examples/streaming/StateMachineExample.jar をそのパスに変更してください。パスのフォーマットは oss://<yourBucketName>/<path>.jar です。

方法 1：ACK コンソールを使用

1. EMR on ACK コンソールにログインします。

2. [EMR on ACK] ページで、対象のクラスターを見つけ、[ACK クラスター] 列のリンクをクリックします。

3. [Pods] ページで、右上隅にある [YAML から作成] をクリックします。

4. create ページで、[サンプルテンプレート] ドロップダウンリストから カスタム を選択し、次の内容をエディターに貼り付けてから、create をクリックします。

   apiVersion: flink.apache.org/v1beta1
   kind: FlinkDeployment
   metadata:
     name: basic-emr-example
   spec:
     flinkVersion: v1_13
     flinkConfiguration:
       taskmanager.numberOfTaskSlots: "2"
       state.savepoints.dir: file:///flink-data/flink-savepoints
       state.checkpoints.dir: file:///flink-data/flink-checkpoints
     serviceAccount: flink
     podTemplate:
       spec:
         serviceAccount: flink
         containers:
           - name: flink-main-container
             volumeMounts:
               - mountPath: /flink-data
                 name: flink-volume
         volumes:
           - name: flink-volume
             emptyDir: {}
   
     jobManager:
       replicas: 1
       resource:
         memory: "2048m"
         cpu: 1
     taskManager:
       resource:
         memory: "2048m"
         cpu: 1
   
     job:
       jarURI: local:///opt/flink/examples/streaming/StateMachineExample.jar
       parallelism: 2
       upgradeMode: stateless

   説明

   この例では Flink 1.13 を使用します。1.13 以外の Flink バージョンを使用する場合は、flinkVersion パラメーターをコンソールに表示されているバージョンに更新してください。

方法 2：kubectl を使用

1. kubectl を使用して Kubernetes クラスターに接続します。詳細については、「ACK クラスターの kubeconfig を取得し、kubectl を使用してクラスターに接続する」をご参照ください。

   API などの他の方法を使用して Kubernetes クラスターに接続することもできます。詳細については、「Kubernetes API の使用」をご参照ください。

2. 次の内容で basic-emr-example.yaml という名前のファイルを作成します。

   apiVersion: flink.apache.org/v1beta1
   kind: FlinkDeployment
   metadata:
     name: basic-emr-example
   spec:
     flinkVersion: v1_13
     flinkConfiguration:
       taskmanager.numberOfTaskSlots: "2"
       state.savepoints.dir: file:///flink-data/flink-savepoints
       state.checkpoints.dir: file:///flink-data/flink-checkpoints
     serviceAccount: flink
     podTemplate:
       spec:
         serviceAccount: flink
         containers:
           - name: flink-main-container
             volumeMounts:
               - mountPath: /flink-data
                 name: flink-volume
         volumes:
           - name: flink-volume
             emptyDir: {}
   
     jobManager:
       replicas: 1
       resource:
         memory: "2048m"
         cpu: 1
     taskManager:
       resource:
         memory: "2048m"
         cpu: 1
   
     job:
       jarURI: local:///opt/flink/examples/streaming/StateMachineExample.jar
       parallelism: 2
       upgradeMode: stateless

   説明

   • ファイル名はカスタマイズできます。この例では、basic-emr-example.yaml を使用します。

   • この例では Flink 1.13 を使用します。異なるバージョンを使用する場合は、flinkVersion パラメーターを Flink のバージョンに合わせて更新してください。

3. 次のコマンドを実行してジョブを送信します。

   kubectl apply -f basic-emr-example.yaml -namespace <your_cluster_namespace>

   説明

   <your_cluster_namespace> をクラスターの名前空間に置き換えてください。名前空間は ACK コンソールの [クラスター詳細] ページで確認できます。