Composant Flink facultatif de Managed Service pour Apache Spark

Le composant facultatif Apache Flink vous permet de créer un cluster Managed Service pour Apache Spark configuré pour exécuter des charges de travail de traitement par lot et de streaming distribuées. L'activation de ce composant prépare l'environnement d'exécution à exécuter des pipelines avec état et des applications de données à grande échelle.

Vous pouvez utiliser votre cluster Flink pour exécuter des jobs Flink à l'aide de la ressource Jobs depuis la console Google Cloud , Google Cloud CLI ou l'API Dataproc, ou exécuter des charges de travail avec la CLI flink sur le nœud maître. Le composant est également compatible avec les jobs Apache Beam et l'exécution sur des clusters Kerberized.

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

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

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

Console

Pour envoyer un exemple de job Flink wordcount à partir de la console :

  1. Ouvrez la page Envoyer un job de Managed Service pour Apache Spark dans la consoleGoogle Cloud de votre navigateur.

  2. Renseignez les champs de la page Envoyer un job :

    1. Sélectionnez le nom du cluster dans la liste des clusters.
    2. Définissez le type de mission sur Flink.
    3. Définissez le champ Main class or jar (Classe principale ou fichier JAR) sur org.apache.flink.examples.java.wordcount.WordCount.
    4. Définissez Fichiers JAR sur file:///usr/lib/flink/examples/batch/WordCount.jar.
      • file:/// désigne un fichier situé sur le cluster. Managed Service pour Apache Spark a installé WordCount.jar lors de la création du cluster Flink.
      • Ce champ accepte également un chemin d'accès Cloud Storage (gs://BUCKET/JARFILE) ou un chemin d'accès au système de fichiers distribué Hadoop (HDFS) (hdfs://PATH_TO_JAR).

  3. Cliquez sur Envoyer.

    • Les résultats des pilotes de jobs s'affichent sur la page Informations sur le job.
    • Les jobs Flink sont listés sur la page Jobs de Managed Service pour Apache Spark dans la console Google Cloud .
    • Cliquez sur Arrêter ou Supprimer sur la page Jobs ou Informations sur le job pour arrêter ou supprimer un job.

gcloud

Pour envoyer un job Flink à un cluster Managed Service pour Apache Spark Flink, exécutez la commande gcloud CLI gcloud dataproc jobs submit localement dans une fenêtre de terminal ou dans Cloud Shell. 

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

Remarques :

  • CLUSTER_NAME : spécifiez le nom du cluster Managed Service pour Apache Spark Flink auquel envoyer le job.
  • REGION : spécifiez une région Compute Engine où se trouve le cluster.
  • MAIN_CLASS : spécifiez la classe main de votre application Flink, par exemple :
    • org.apache.flink.examples.java.wordcount.WordCount
  • JAR_FILE : spécifiez le fichier JAR de l'application Flink. Vous pouvez spécifier :
    • Un fichier JAR installé sur le cluster, à l'aide du préfixe file:///` :
      • file:///usr/lib/flink/examples/streaming/TopSpeedWindowing.jar
      • file:///usr/lib/flink/examples/batch/WordCount.jar
    • Fichier JAR dans Cloud Storage : gs://BUCKET/JARFILE
    • Un fichier JAR dans HDFS : hdfs://PATH_TO_JAR

  • JOB_ARGS : vous pouvez ajouter des arguments de job après le double tiret (--).

  • Une fois le job envoyé, la sortie du pilote de job s'affiche dans le terminal local ou Cloud Shell.

    Program execution finished
Job with JobID 829d48df4ebef2817f4000dfba126e0f has finished.
Job Runtime: 13610 ms
...
(after,1)
(and,12)
(arrows,1)
(ay,1)
(be,4)
(bourn,1)
(cast,1)
(coil,1)
(come,1)

REST

Cette section explique comment envoyer un job Flink à un cluster Managed Service pour Apache Spark Flink à l'aide de l'API jobs.submit de Managed Service pour Apache Spark.

Avant d'utiliser les données de requête, effectuez les remplacements suivants :

  • PROJECT_ID : Google Cloud ID du projet
  • REGION : région du cluster
  • CLUSTER_NAME : spécifiez le nom du cluster Managed Service pour Apache Spark Flink auquel envoyer le job.

Méthode HTTP et URL : 

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

Corps JSON de la requête :

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

Pour envoyer votre requête, développez l'une des options suivantes :

Vous devriez recevoir une réponse JSON de ce type :

{
  "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"
}
  • Les jobs Flink sont listés sur la page Jobs de Managed Service pour Apache Spark dans la console Google Cloud .
  • Vous pouvez cliquer sur Arrêter ou Supprimer sur la page Jobs (Jobs) ou Job details (Détails du job) de la console Google Cloud pour arrêter ou supprimer un job.

Au lieu d'exécuter des jobs Flink à l'aide de la ressource Managed Service pour Apache Spark Jobs, vous pouvez exécuter des jobs Flink sur le nœud pilote de votre cluster Flink à l'aide de la CLI flink.

Les sections suivantes décrivent différentes manières d'exécuter un job CLI flink sur votre cluster Managed Service pour Apache Spark Flink.

  1. Se connecter au nœud maître via SSH : utilisez l'utilitaire SSH pour ouvrir une fenêtre de terminal sur la VM maître du cluster.

  2. Définissez le chemin d'accès aux classes : initialisez le chemin d'accès aux classes Hadoop à partir de la fenêtre du terminal SSH sur la VM maître 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 : application, par job et session.

    1. Mode Application : le mode Application Flink est compatible avec la version 2.0 et les versions ultérieures de l'image Managed Service pour Apache Spark. Ce mode exécute la méthode main() du job sur le gestionnaire de jobs YARN. Le cluster s'arrête une fois le job terminé.

      Exemple d'envoi de job :

      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épertoriez les jobs en cours d'exécution :

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

      Annuler un job en cours d'exécution :

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

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

      Exemple d'envoi de job :

      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 un ou plusieurs jobs à 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 cette option est activée, une fois le cluster créé, Managed Service pour Apache Spark 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 à l'aide de la commande yarn application -list.

        2. Exécutez le script Flink yarn-session.sh, préinstallé sur la VM maître de 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 de wrapper Flink /usr/bin/flink-yarn-daemon avec les paramètres par défaut :

          . /usr/bin/flink-yarn-daemon

      2. Soumettre un travail à une session : Exécutez la commande suivante pour soumettre un travail 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 où 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. Répertorier les jobs d'une session : pour répertorier les jobs Flink d'une session, effectuez l'une des opérations 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 de la sortie 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 de la sortie 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

Vous pouvez exécuter des jobs Apache Beam sur Managed Service pour Apache Spark à l'aide de 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 un job Java Beam à partir du nœud maître du cluster Managed Service pour Apache Spark.

  1. Créez un cluster Managed Service pour Apache Spark 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 Managed Service pour Apache Spark 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 la section Bonnes pratiques concernant les champs d'application). Le champ d'application cloud-platform est activé par défaut (vous n'avez pas besoin d'inclure ce paramètre d'option) lorsque vous créez un cluster qui utilise la version 2.1 ou ultérieure de l'image Managed Service pour Apache Spark.

  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 Flink YARN sur le nœud maître du cluster Managed Service pour Apache Spark.

    . /usr/bin/flink-yarn-daemon

    Notez la version de Flink sur votre cluster Managed Service pour Apache Spark.

    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 sur votre cluster Managed Service pour Apache Spark. Consultez le tableau Compatibilité des versions de Flink qui liste 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 Managed Service pour Apache Spark. Vous pouvez utiliser gcloud storage cp pour accélérer les transferts de fichiers vers votre cluster Managed Service pour Apache Spark depuis 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 votre Managed Service pour Apache Spark, téléchargez le fichier uber JAR.

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

  7. Exécutez le job Java Beam sur le nœud pilote du cluster Managed Service pour Apache Spark.

    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 Flink Runner de Beam (voir également la feuille de route du framework de portabilité).

L'exemple suivant exécute un job Beam portable en Python à partir du nœud maître du cluster Managed Service pour Apache Spark.

  1. Créez un cluster Managed Service pour Apache Spark 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 : version d'image du cluster, qui détermine la version de Flink installée sur le cluster (par exemple, consultez les versions du composant Apache Flink listées pour la dernière version et les quatre versions précédentes de l'image 2.0.x).
    • --region : région Managed Service pour Apache Spark disponible.
    • --enable-component-gateway : activez l'accès à l'interface utilisateur du gestionnaire de jobs Flink.
    • --scopes : activez l'accès aux API Google Cloud par votre cluster (voir Bonnes pratiques concernant les champs d'application). Le champ d'application cloud-platform est activé par défaut (vous n'avez pas besoin d'inclure ce paramètre d'option) lorsque vous créez un cluster qui utilise la version 2.1 ou ultérieure de l'image Managed Service pour Apache Spark.

  2. Utilisez gcloud CLI localement ou dans Cloud Shell pour créer un bucket Cloud Storage. Vous spécifierez le 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, l'adresse du maître Flink où les jobs sont exécutés. Vous spécifierez 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 qui exécute le cluster Managed Service pour Apache Spark. Vous spécifierez FLINK_VERSION lorsque vous exécuterez un exemple de programme de comptage de mots.

    flink --version

  4. Installez les bibliothèques Python nécessaires au job sur le nœud maître du cluster.

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

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

  6. Exécutez l'exemple de décompte de mots sur le maître de 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, indiqué précédemment.
    • --flink_master : FLINK_MASTER_URL, indiqué 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 é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

Le composant Flink de Managed Service pour Apache Spark est compatible avec les clusters Kerberized. Un ticket Kerberos valide est nécessaire pour envoyer et conserver un job Flink ou pour démarrer un cluster Flink. Par défaut, un ticket Kerberos reste valide pendant sept jours.

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éer un cluster Managed Service pour Apache Spark Flink
  2. Après avoir créé le cluster, cliquez sur le lien YARN ResourceManager 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. Selon l'état d'achèvement d'un job, un lien ApplicationMaster ou Historique s'affiche.
  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.