Composant Flink facultatif de Dataproc

Vous pouvez activer des composants supplémentaires tels que Flink lorsque vous créez un cluster Dataproc à l'aide de la fonctionnalité Composants facultatifs. Cette page vous explique comment créer un cluster Dataproc avec le composant facultatif Apache Flink activé (un cluster Flink), puis comment exécuter des jobs Flink sur le cluster.

Vous pouvez utiliser votre cluster Flink pour :

1. Exécutez des jobs Flink à l'aide de la ressource Jobs Dataproc depuis la console Google Cloud , la Google Cloud CLI ou l'API Dataproc.

2. Exécutez des jobs Flink à l'aide de la CLI flink sur le nœud maître du cluster Flink.

3. Exécuter des jobs Apache Beam sur Flink

4. Exécuter Flink sur un cluster kerberisé

Créer un cluster Dataproc Flink

Vous pouvez utiliser la console Google Cloud , Google Cloud CLI ou l'API Dataproc pour créer un cluster Dataproc sur lequel le composant Flink est activé.

Recommandation : Utilisez un cluster de VM standard à un nœud maître avec le composant Flink. Les clusters Dataproc en mode haute disponibilité (avec trois VM maîtres) ne sont pas compatibles avec le mode haute disponibilité Flink.

gcloud dataproc clusters create CLUSTER_NAME \
    --region=REGION \
    --image-version=DATAPROC_IMAGE_VERSION \
    --optional-components=FLINK \
    --enable-component-gateway \
    --properties=PROPERTIES
    ... other flags

Après avoir créé un cluster Flink

• Utilisez le lien Flink History Server dans la passerelle des composants pour afficher le serveur d'historique Flink exécuté sur le cluster Flink.
• Utilisez YARN ResourceManager link dans la passerelle de composants pour afficher l'interface Web du gestionnaire de tâches Flink exécutée sur le cluster Flink .
• Créez un serveur d'historique persistant Dataproc pour afficher les fichiers d'historique des tâches Flink écrits par des clusters Flink existants et supprimés.

Exécuter des jobs Flink à l'aide de la ressource Dataproc Jobs

Vous pouvez exécuter des jobs Flink à l'aide de la ressource Jobs Dataproc depuis la consoleGoogle Cloud , Google Cloud CLI ou l'API Dataproc.

gcloud dataproc jobs submit flink \
    --cluster=CLUSTER_NAME \
    --region=REGION \
    --class=MAIN_CLASS \
    --jar=JAR_FILE \
    -- JOB_ARGS

POST https://dataproc.googleapis.com/v1/projects/PROJECT_ID/regions/REGION/jobs:submit

{
  "job": {
    "placement": {
      "clusterName": "CLUSTER_NAME"
    },
    "flinkJob": {
      "mainClass": "org.apache.flink.examples.java.wordcount.WordCount",
      "jarFileUris": [
        "file:///usr/lib/flink/examples/batch/WordCount.jar"
      ]
    }
  }
}

curl -X POST \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://dataproc.googleapis.com/v1/projects/PROJECT_ID/regions/REGION/jobs:submit"

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
    -Method POST `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://dataproc.googleapis.com/v1/projects/PROJECT_ID/regions/REGION/jobs:submit" | Select-Object -Expand Content

{
  "reference": {
    "projectId": "PROJECT_ID",
    "jobId": "JOB_ID"
  },
  "placement": {
    "clusterName": "CLUSTER_NAME",
    "clusterUuid": "CLUSTER_UUID"
  },
  "flinkJob": {
    "mainClass": "org.apache.flink.examples.java.wordcount.WordCount",
    "args": [
      "1000"
    ],
    "jarFileUris": [
      "file:///usr/lib/flink/examples/batch/WordCount.jar"
    ]
  },
  "status": {
    "state": "PENDING",
    "stateStartTime": "2020-10-07T20:16:21.759Z"
  },
  "jobUuid": "JOB_UUID"
}

Exécuter des jobs Flink à l'aide de la CLI flink

Au lieu d'exécuter des tâches Flink à l'aide de la ressource Dataproc Jobs, vous pouvez exécuter des tâches Flink sur le nœud maître de votre cluster Flink à l'aide de la CLI flink.

Les sections suivantes décrivent différentes manières d'exécuter un job de CLI flink sur votre cluster Dataproc Flink.

1. Connectez-vous en SSH au nœud maître : utilisez l'utilitaire SSH pour ouvrir une fenêtre de terminal sur la VM maître du cluster.

2. Définissez le chemin de classe : initialisez le chemin de classe Hadoop à partir de la fenêtre de terminal SSH sur la VM maîtresse du cluster Flink :

   export HADOOP_CLASSPATH=$(hadoop classpath)
   

3. Exécuter des jobs Flink : vous pouvez exécuter des jobs Flink dans différents modes de déploiement sur YARN : mode application, mode par job et mode session.

   1. Mode Application : le mode Application Flink est compatible avec la version 2.0 et les versions ultérieures de l'image Dataproc. Dans ce mode, la méthode main() du job est exécutée sur le gestionnaire de jobs YARN. Le cluster s'arrête une fois le job terminé.

      Exemple d'envoi de tâche :

      flink run-application \
          -t yarn-application \
          -Djobmanager.memory.process.size=1024m \
          -Dtaskmanager.memory.process.size=2048m \
          -Djobmanager.heap.mb=820 \
          -Dtaskmanager.heap.mb=1640 \
          -Dtaskmanager.numberOfTaskSlots=2 \
          -Dparallelism.default=4 \
          /usr/lib/flink/examples/batch/WordCount.jar
      

      Répertorier les jobs en cours d'exécution :

      ./bin/flink list -t yarn-application -Dyarn.application.id=application_XXXX_YY
      

      Annuler une tâche en cours d'exécution :

      ./bin/flink cancel -t yarn-application -Dyarn.application.id=application_XXXX_YY <jobId>
      

   2. Mode par job : ce mode Flink exécute la méthode main() du job côté client.

      Exemple d'envoi de tâche :

      flink run \
          -m yarn-cluster \
          -p 4 \
          -ys 2 \
          -yjm 1024m \
          -ytm 2048m \
          /usr/lib/flink/examples/batch/WordCount.jar
      

   3. Mode session : démarrez une session Flink YARN de longue durée, puis envoyez une ou plusieurs tâches à la session.

      1. Démarrer une session : vous pouvez démarrer une session Flink de l'une des manières suivantes :

         1. Créez un cluster Flink en ajoutant l'option --metadata flink-start-yarn-session=true à la commande gcloud dataproc clusters create (consultez Créer un cluster Dataproc Flink). Lorsque cet indicateur est activé, une fois le cluster créé, Dataproc exécute /usr/bin/flink-yarn-daemon pour démarrer une session Flink sur le cluster.

            L'ID d'application YARN de la session est enregistré dans /tmp/.yarn-properties-${USER}. Vous pouvez lister l'ID avec la commande yarn application -list.

         2. Exécutez le script Flink yarn-session.sh, qui est préinstallé sur la VM maître du cluster, avec des paramètres personnalisés :

            Exemple avec des paramètres personnalisés :

            /usr/lib/flink/bin/yarn-session.sh \
                -s 1 \
                -jm 1024m \
                -tm 2048m \
                -nm flink-dataproc \
                --detached
            

         3. Exécutez le script wrapper Flink /usr/bin/flink-yarn-daemon avec les paramètres par défaut :

            . /usr/bin/flink-yarn-daemon
            

      2. Envoyer un job à une session : exécutez la commande suivante pour envoyer un job Flink à la session.

         flink run -m <var>FLINK_MASTER_URL</var>/usr/lib/flink/examples/batch/WordCount.jar
         

         • FLINK_MASTER_URL : URL, y compris l'hôte et le port, de la VM maître Flink sur laquelle les jobs sont exécutés. Supprimez http:// prefix de l'URL. Cette URL est indiquée dans le résultat de la commande lorsque vous démarrez une session Flink. Vous pouvez exécuter la commande suivante pour lister cette URL dans le champ Tracking-URL :

         yarn application -list -appId=<yarn-app-id> | sed 's#http://##'
            ```
         

      3. Lister les jobs dans une session : pour lister les jobs Flink dans une session, procédez de l'une des manières suivantes :

         • Exécutez flink list sans arguments. La commande recherche l'ID d'application YARN de la session dans /tmp/.yarn-properties-${USER}.

         • Obtenez l'ID application YARN de la session à partir de /tmp/.yarn-properties-${USER} ou du résultat de yarn application -list, puis exécutez <code>flink list -yid YARN_APPLICATION_ID.

         • Exécutez flink list -m FLINK_MASTER_URL.

      4. Arrêter une session : pour arrêter la session, obtenez l'ID application YARN de la session à partir de /tmp/.yarn-properties-${USER} ou du résultat de yarn application -list, puis exécutez l'une des commandes suivantes :

         echo "stop" | /usr/lib/flink/bin/yarn-session.sh -id YARN_APPLICATION_ID
         

         yarn application -kill YARN_APPLICATION_ID
         

Exécuter des jobs Apache Beam sur Flink

Vous pouvez exécuter des tâches Apache Beam sur Dataproc à l'aide de la commande FlinkRunner.

Vous pouvez exécuter des tâches Beam sur Flink de différentes manières:

1. Tâches Java Beam
2. Tâches portables Beam

Tâches Java Beam

Empaquetez vos tâches Beam dans un fichier JAR. Fournissez le fichier JAR groupé avec les dépendances nécessaires à l'exécution de la tâche.

L'exemple suivant exécute une tâche Java Beam à partir du nœud maître du cluster Dataproc.

1. Créez un cluster Dataproc avec le composant Flink activé.

   gcloud dataproc clusters create CLUSTER_NAME \
       --optional-components=FLINK \
       --image-version=DATAPROC_IMAGE_VERSION \
       --region=REGION \
       --enable-component-gateway \
       --scopes=https://www.googleapis.com/auth/cloud-platform
   

   • --optional-components : Flink
   • --image-version : La version de l'image du cluster qui détermine la version Flink installée sur le cluster (par exemple, consultez les versions des composants Apache Flink répertoriées pour les quatre versions suivantes et précédentes des Versions 2.0.x de l'image).
   • --region: région Dataproc compatible
   • --enable-component-gateway: permet d'accéder à l'interface utilisateur du gestionnaire de tâches Flink.
   • --scopes : activez l'accès aux API Google Cloud par votre cluster (consultez les bonnes pratiques concernant les niveaux d'accès). Le champ d'application cloud-platform est activé par défaut (vous n'avez pas besoin d'inclure ce paramètre d'indicateur) lorsque vous créez un cluster qui utilise la version d'image Dataproc 2.1 ou ultérieure.

2. Utilisez l'utilitaire SSH pour ouvrir une fenêtre de terminal sur le nœud maître du cluster Flink.

3. Démarrez une session YARN Flink sur le nœud maître du cluster Dataproc.

   . /usr/bin/flink-yarn-daemon
   

   Notez la version de Flink sur votre cluster Dataproc.

   flink --version
   

4. Sur votre ordinateur local, générez l'exemple canonique de nombre de mots Beam en Java.

   Choisissez une version de Beam compatible avec la version de Flink de votre cluster Dataproc. Consultez le tableau Compatibilité avec les versions de Flink qui répertorie la compatibilité des versions de Beam-Flink.

   Ouvrez le fichier POM généré. Vérifiez la version de l'exécuteur Beam Flink spécifiée par le tag <flink.artifact.name>. Si la version de l'exécuteur Beam Flink dans le nom de l'artefact Flink ne correspond pas à la version Flink de votre cluster, mettez à jour le numéro de version.

   mvn archetype:generate \
       -DarchetypeGroupId=org.apache.beam \
       -DarchetypeArtifactId=beam-sdks-java-maven-archetypes-examples \
       -DarchetypeVersion=BEAM_VERSION \
       -DgroupId=org.example \
       -DartifactId=word-count-beam \
       -Dversion="0.1" \
       -Dpackage=org.apache.beam.examples \
       -DinteractiveMode=false
   

5. Empaqueter l'exemple de décompte de mots.

   mvn package -Pflink-runner
   

6. Importez le fichier uber JAR empaqueté word-count-beam-bundled-0.1.jar (~135 Mo) sur le nœud maître de votre cluster Dataproc. Vous pouvez utiliser gcloud storage cp pour accélérer les transferts de fichiers vers votre cluster Dataproc à partir de Cloud Storage.

   1. Sur votre terminal local, créez un bucket Cloud Storage et importez le fichier Uber JAR.

      gcloud storage buckets create BUCKET_NAME
      
      gcloud storage cp target/word-count-beam-bundled-0.1.jar gs://BUCKET_NAME/
      

   2. Sur le nœud maître de Dataproc, téléchargez le fichier Uber JAR.

      gcloud storage cp gs://BUCKET_NAME/word-count-beam-bundled-0.1.jar .
      

7. Exécutez la tâche Java Beam sur le nœud maître du cluster Dataproc.

   flink run -c org.apache.beam.examples.WordCount word-count-beam-bundled-0.1.jar \
       --runner=FlinkRunner \
       --output=gs://BUCKET_NAME/java-wordcount-out
   

8. Vérifiez que les résultats ont bien été écrits dans votre bucket Cloud Storage.

   gcloud storage cat gs://BUCKET_NAME/java-wordcount-out-SHARD_ID
   

9. Arrêtez la session YARN Flink.

   yarn application -list
   
   yarn application -kill YARN_APPLICATION_ID
   

Tâches Beam portables

Pour exécuter des jobs Beam écrits en Python, Go et d'autres langages compatibles, vous pouvez utiliser FlinkRunner et PortableRunner, comme décrit sur la page de l'exécuteur Flink de Beam (voir aussi Feuille de route du framework de portabilité).

L'exemple suivant exécute une tâche Beam portable en Python à partir du nœud maître du cluster Dataproc.

1. Créez un cluster Dataproc avec les composants Flink et Docker activés.

   gcloud dataproc clusters create CLUSTER_NAME \
       --optional-components=FLINK,DOCKER \
       --image-version=DATAPROC_IMAGE_VERSION \
       --region=REGION \
       --enable-component-gateway \
       --scopes=https://www.googleapis.com/auth/cloud-platform
   

   Remarques :

   • --optional-components: Flink et Docker.
   • --image-version : la version de l'image du cluster, qui détermine la version Flink installée sur le cluster (par exemple, consultez les versions des composants Apache Flink répertoriées pour les quatre versions suivantes et précédentes des versions 2.0.x de l'image).
   • --region : région Dataproc disponible
   • --enable-component-gateway : permet d'accéder à l'interface utilisateur du gestionnaire de tâches Flink.
   • --scopes : activez l'accès aux API Google Cloud par votre cluster (voir Bonnes pratiques concernant les niveaux d'accès). Le champ d'application cloud-platform est activé par défaut (vous n'avez pas besoin d'inclure ce paramètre d'indicateur) lorsque vous créez un cluster qui utilise la version d'image Dataproc 2.1 ou ultérieure.

2. Utilisez gcloud CLI en local ou dans Cloud Shell pour créer un bucket Cloud Storage. Vous spécifierez BUCKET_NAME lorsque vous exécuterez un exemple de programme de comptage de mots.

   gcloud storage buckets create BUCKET_NAME
   

3. Dans une fenêtre de terminal sur la VM du cluster, démarrez une session Flink YARN. Notez l'URL du maître Flink, c'est-à-dire l'adresse du maître Flink où les jobs sont exécutés. Vous spécifierez le FLINK_MASTER_URL lorsque vous exécuterez un exemple de programme de comptage de mots.

   . /usr/bin/flink-yarn-daemon
   

   Affichez et notez la version de Flink exécutée par le cluster Dataproc. Vous spécifierez le FLINK_VERSION lorsque vous exécuterez un exemple de programme de comptage de mots.

   flink --version
   

4. Installez les bibliothèques Python nécessaires pour exécuter la tâche sur le nœud maître du cluster.

5. Installez une version de Beam compatible avec la version de Flink du cluster.

   python -m pip install apache-beam[gcp]==BEAM_VERSION
   

6. Exécutez l'exemple de comptage de mots sur le nœud maître du cluster.

   python -m apache_beam.examples.wordcount \
       --runner=FlinkRunner \
       --flink_version=FLINK_VERSION \
       --flink_master=FLINK_MASTER_URL
       --flink_submit_uber_jar \
       --output=gs://BUCKET_NAME/python-wordcount-out
   

   Remarques :

   • --runner : FlinkRunner.
   • --flink_version : FLINK_VERSION, noté précédemment.
   • --flink_master : FLINK_MASTER_URL, noté précédemment.
   • --flink_submit_uber_jar : utilisez le fichier Uber JAR pour exécuter le job Beam.
   • --output : BUCKET_NAME, créé précédemment.

7. Vérifiez que les résultats ont bien été écrits dans votre bucket.

   gcloud storage cat gs://BUCKET_NAME/python-wordcount-out-SHARD_ID
   

8. Arrêtez la session YARN Flink.

   1. Obtenez l'ID application.

   yarn application -list
   
   1. Insert the <var>YARN_APPLICATION_ID</var>, then stop the session.
   
   yarn application -kill 
   

Exécuter Flink sur un cluster kerberisé

Le composant Dataproc Flink est compatible avec les clusters kerberisés. Une demande Kerberos valide est nécessaire pour envoyer et conserver une tâche Flink, ou pour démarrer un cluster Flink. Par défaut, une demande Kerberos reste valide pendant sept jours.

Accéder à l'interface utilisateur du gestionnaire de tâches Flink

L'interface Web du gestionnaire de tâches Flink est disponible lorsqu'une tâche Flink ou un cluster de session Flink est en cours d'exécution. Pour utiliser l'interface Web :

1. Créez un cluster Dataproc Flink.
2. Après la création du cluster, cliquez sur le lien du ResourceManager YARN de la passerelle des composants dans l'onglet "Interface Web" de la page Détails du cluster de la console Google Cloud .
3. Dans l'interface utilisateur YARN Resource Manager, identifiez l'entrée de l'application de cluster Flink. En fonction de l'état d'exécution d'un job, un lien ApplicationMaster ou History (Historique) est répertorié.

   

4. Pour une tâche de streaming de longue durée, cliquez sur le lien ApplicationManager pour ouvrir le tableau de bord Flink. Pour une tâche terminée, cliquez sur le lien Historique pour afficher les détails de la tâche.