E-MapReduce:Flink ジョブの送信

前提条件

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

注意事項

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

その後、設定内の local:///opt/flink/examples/streaming/StateMachineExample.jar を、Alibaba Cloud OSS 内の JAR ファイルへのパスに置き換える必要があります。パスは oss://<yourBucketName>/<path>.jar の形式を使用する必要があります。

方法 1：ACK コンソールを使用したジョブの送信

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

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

3. ポッド ページで、右上隅の 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 を使用します。異なる Flink バージョンを使用する場合は、flinkVersion パラメーターを適宜更新してください。

方法 2：kubectl を使用したジョブの送信

1. kubectl を使用して Kubernetes クラスターに接続します。 詳細については、「クラスターの 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 を使用します。異なる Flink バージョンを使用する場合は、flinkVersion パラメーターを適宜更新してください。

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

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

   説明

   <cluster-namespace> を、お使いのクラスターの名前空間に置き換えます。名前空間は、EMR on ACK コンソールのクラスター管理ページで確認できます。