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 explique comment créer un cluster Dataproc avec le kit Apache Flink un composant facultatif activé (cluster Flink), puis exécuter des jobs Flink sur le cluster.
Vous pouvez utiliser votre cluster Flink pour :
Exécuter des jobs Flink à l'aide de la ressource Dataproc
Jobs
depuis la console Google Cloud, Google Cloud CLI ou l'API Dataproc.Exécutez des tâches Flink à l'aide de la CLI
flink
exécutée sur le nœud maître du cluster Flink.Exécuter Flink sur un cluster kerberisé
Créer un cluster Dataproc Flink
Vous pouvez utiliser la console Google Cloud, la 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 à un maître standard avec le composant Flink. Clusters en mode haute disponibilité Dataproc (avec trois VM maîtres) ne sont pas compatibles Mode haute disponibilité Flink :
Console
Pour créer un cluster Dataproc Flink à l'aide de la console Google Cloud, procédez comme suit:
Ouvrir Dataproc Page Créer un cluster Dataproc sur Compute Engine.
- Le panneau Configurer un cluster est sélectionné.
- Dans la section Gestion des versions, confirmez ou modifiez le paramètre
Type et version de l'image. La version de l'image du cluster détermine la version du composant Flink installé sur le cluster.
- Pour activer le composant Flink sur le cluster, vous devez disposer de la version d'image 1.5 ou ultérieure Consultez la page Versions compatibles de Dataproc. pour afficher la liste des versions des composants incluses dans chaque la version de l'image Dataproc).
- La version de l'image doit être [TBD] ou ultérieure pour exécuter des tâches Flink via l'API Dataproc Jobs (voir Exécuter des tâches Flink Dataproc).
- Dans la section Composants:
- Sous Component Gateway (Passerelle des composants), sélectionnez Activez la passerelle des composants. Vous devez activer le paramètre Passerelle des composants pour activer le lien de la passerelle des composants vers l'interface utilisateur du serveur d'historique Flink. L'activation de la passerelle des composants active également un accès à l'interface Web de Flink Job Manager ; en cours d'exécution sur le cluster Flink.
- Sous Composants facultatifs, sélectionnez Flink et autres options facultatives à activer sur votre cluster.
- Dans la section Gestion des versions, confirmez ou modifiez le paramètre
Type et version de l'image. La version de l'image du cluster détermine la version du composant Flink installé sur le cluster.
Cliquez sur le panneau Personnaliser le cluster (facultatif).
Dans la section Propriétés du cluster, cliquez sur Ajouter des propriétés pour chaque option propriété du cluster à ajouter à votre cluster. Vous pouvez ajouter des propriétés avec un préfixe
flink
pour configurer des propriétés Flink dans/etc/flink/conf/flink-conf.yaml
qui serviront de valeurs par défaut pour les applications Flink que vous exécutez sur le cluster.Exemples :
- Définissez
flink:historyserver.archive.fs.dir
pour spécifier l'emplacement Cloud Storage dans lequel écrire les fichiers d'historique des tâches Flink (cet emplacement sera utilisé par le serveur d'historique Flink exécuté sur le cluster Flink). - Définissez les emplacements de tâche Flink avec
flink:taskmanager.numberOfTaskSlots=n
.
- Définissez
Dans la section Métadonnées de cluster personnalisées, cliquez sur Ajouter des métadonnées pour ajouter et facultatives. Par exemple, ajoutez
flink-start-yarn-session
true
pour exécuter le daemon Flink YARN (/usr/bin/flink-yarn-daemon
) en arrière-plan sur le maître de cluster pour démarrer une session Flink YARN (voir Mode de session Flink).
Si vous utilisez la version 2.0 ou une version antérieure de l'image Dataproc, Cliquez sur le panneau Gérer la sécurité (facultatif), puis sous Accès au projet : sélectionnez
Enables the cloud-platform scope for this cluster
. Le champ d'applicationcloud-platform
est activé par défaut lorsque vous créez un cluster qui utilise la version 2.1 ou ultérieure de l'image Dataproc.
- Le panneau Configurer un cluster est sélectionné.
Cliquez sur Créer pour créer le cluster.
gcloud
Pour créer un cluster Dataproc Flink à l'aide de la gcloud CLI, exécutez la commande suivante : gcloud dataproc clusters create localement dans une fenêtre de terminal ou dans Cloud Shell:
gcloud dataproc clusters create CLUSTER_NAME \ --region=REGION \ --image-version=DATAPROC_IMAGE_VERSION \ --optional-components=FLINK \ --enable-component-gateway \ --properties=PROPERTIES ... other flags
Remarques :
- CLUSTER_NAME: spécifiez le nom du cluster.
- REGION: spécifiez une région Compute Engine. l'emplacement du cluster.
DATAPROC_IMAGE_VERSION : spécifiez éventuellement la version de l'image à utiliser sur le cluster. La version de l'image du cluster détermine du composant Flink installé sur le cluster.
Pour activer le composant Flink sur le cluster, vous devez disposer de la version d'image 1.5 ou ultérieure Consultez la page Versions compatibles de Dataproc. pour afficher la liste des versions des composants incluses dans chaque la version de l'image Dataproc).
La version de l'image doit être [TBD] ou ultérieure pour exécuter des jobs Flink via l'API Dataproc Jobs (consultez Exécuter des jobs Dataproc Flink).
--optional-components
: vous devez spécifier le composantFLINK
pour exécuter les jobs Flink et le service Web HistoryServer Flink sur le cluster.--enable-component-gateway
: vous devez activer le Passerelle des composants vers activer le lien de la passerelle des composants vers l'interface utilisateur du serveur d'historique Flink. L'activation de la passerelle des composants permet également d'accéder l'interface Web Flink Job Manager s'exécutant sur le cluster Flink.PROPERTIES. Vous pouvez également spécifier une ou plusieurs propriétés de cluster.
Lorsque vous créez des clusters Dataproc avec les versions d'image
2.0.67
et2.1.15
, vous pouvez utiliser l'indicateur--properties
pour configurer des propriétés Flink dans/etc/flink/conf/flink-conf.yaml
qui serviront de valeurs par défaut pour les applications Flink que vous exécutez sur le cluster.Vous pouvez définir
flink:historyserver.archive.fs.dir
pour spécifier l'emplacement Cloud Storage dans lequel écrire l'historique des jobs Flink (cet emplacement sera utilisé par le serveur d'historique Flink qui exécute sur le cluster Flink).Exemple de propriétés multiples:
--properties=flink:historyserver.archive.fs.dir=gs://my-bucket/my-flink-cluster/completed-jobs,flink:taskmanager.numberOfTaskSlots=2
Autres indicateurs:
- Vous pouvez ajouter le
--metadata flink-start-yarn-session=true
facultatif pour exécuter le daemon Flink YARN (/usr/bin/flink-yarn-daemon
) en arrière-plan sur le nœud maître du cluster pour démarrer une session Flink YARN (voir Mode de session Flink).
- Vous pouvez ajouter le
Si vous utilisez une version d'image 2.0 ou antérieure, vous pouvez ajouter le paramètre
--scopes=https://www.googleapis.com/auth/cloud-platform
sur activer l'accès aux API Google Cloud depuis votre cluster (voir Bonnes pratiques concernant les niveaux d'accès). Le champ d'applicationcloud-platform
est activé par défaut lorsque vous créez un cluster qui utilise la version 2.1 ou ultérieure de l'image Dataproc.
API
Pour créer un cluster Dataproc Flink à l'aide de l'API Dataproc, envoyez une clusters.create comme suit:
Remarques :
Définissez SoftwareConfig.Component sur
FLINK
.Vous pouvez éventuellement définir
SoftwareConfig.imageVersion
pour spécifier la version d'image à utiliser sur le cluster. La version de l'image du cluster détermine du composant Flink installé sur le cluster.La version de l'image doit être 1.5 ou ultérieure pour activer le composant Flink sur le cluster (consultez la section Versions Dataproc compatibles pour afficher la liste des versions de composant incluses dans chaque version d'image Dataproc).
La version de l'image doit être [TBD] ou ultérieure pour exécuter des tâches Flink via l'API Dataproc Jobs (voir Exécuter des tâches Flink Dataproc).
Définir EndpointConfig.enableHttpPortAccess sur
true
pour activer la passerelle des composants vers l'UI du serveur d'historique Flink. L'activation de la passerelle des composants permet également d'accéder l'interface Web Flink Job Manager s'exécutant sur le cluster Flink.Vous pouvez éventuellement définir
SoftwareConfig.properties
pour spécifier une ou plusieurs propriétés de cluster.- Vous pouvez spécifier des propriétés Flink qui agiront en tant que
par défaut pour les applications Flink que vous exécutez sur le cluster. Par exemple :
vous pouvez définir
flink:historyserver.archive.fs.dir
pour spécifier l'emplacement Cloud Storage dans lequel écrire l'historique des jobs Flink (cet emplacement sera utilisé par le serveur d'historique Flink qui exécute sur le cluster Flink).
- Vous pouvez spécifier des propriétés Flink qui agiront en tant que
par défaut pour les applications Flink que vous exécutez sur le cluster. Par exemple :
vous pouvez définir
Vous pouvez éventuellement définir les éléments suivants:
GceClusterConfig.metadata
Par exemple, pour spécifierflink-start-yarn-session
true
afin d'exécuter le daemon Flink YARN (/usr/bin/flink-yarn-daemon
) en arrière-plan sur le maître de cluster pour démarrer une session Flink YARN (voir Mode de session Flink).- GceClusterConfig.serviceAccountScopes à
https://www.googleapis.com/auth/cloud-platform
(champ d'applicationcloud-platform
) lorsque vous utilisez des versions d'image 2.0 ou antérieures pour activer l'accès aux API Google Cloud par votre cluster (voir Bonnes pratiques concernant les champs d'application). Le champ d'applicationcloud-platform
est activé par défaut lorsque vous créez un cluster qui utilise la version 2.1 ou ultérieure de l'image Dataproc.
Après avoir créé un cluster Flink
- Utilisez le lien
Flink History Server
dans Passerelle des composants pour afficher le serveur d'historique Flink exécuté sur le cluster Flink. - Utilisez
YARN ResourceManager link
dans la passerelle des 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 Dataproc Jobs
à partir de
la console Google Cloud, la Google Cloud CLI ou l'API Dataproc.
Console
Pour envoyer un exemple de job de décompte de mots Flink à partir de la console:
Ouvrir Dataproc la page Envoyer une tâche dans le la console Google Cloud dans votre navigateur.
Renseignez les champs de la page Envoyer une tâche :
- Sélectionnez le nom du cluster dans la liste des clusters.
- Définissez le champ Job type (Type de job) sur
Flink
. - Définissez le champ Main class or jar (Classe principale ou fichier JAR) sur
org.apache.flink.examples.java.wordcount.WordCount
. - Définissez Fichiers JAR sur
file:///usr/lib/flink/examples/batch/WordCount.jar
.file:///
désigne un fichier situé sur le cluster. Dataproc installéWordCount.jar
lors de la création du cluster Flink.- Ce champ accepte aussi un chemin d'accès Cloud Storage
(
gs://BUCKET/JARFILE
) ou un Chemin d'accès HDFS (Hadoop Distributed File System) (hdfs://PATH_TO_JAR
).
Cliquez sur Envoyer.
- Les résultats du pilote de tâches s'affichent sur la page Informations sur la tâche.
- Les tâches Flink sont répertoriées sur la page Jobs (Tâches) de Dataproc dans la console Google Cloud.
- Cliquez sur Arrêter ou Supprimer sur la page Tâches ou Informations sur la tâche. pour arrêter ou supprimer un job.
gcloud
Pour envoyer un job Flink à un cluster Dataproc Flink, exécutez la 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 Flink Dataproc. auquel envoyer le job.
- REGION: spécifiez une région Compute Engine. l'emplacement du 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 les éléments suivants :
- 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
- Un fichier JAR installé sur le cluster, à l'aide du préfixe
JOB_ARGS: vous pouvez éventuellement ajouter des arguments pour la tâche après le double tiret (
--
).Une fois la tâche envoyée, le résultat du pilote de tâches 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 job Flink Dataproc à l'aide de Dataproc jobs.submit
Avant d'utiliser les données de requête ci-dessous, effectuez les remplacements suivants :
- PROJECT_ID : ID de projet Google Cloud
- REGION : région du cluster
- CLUSTER_NAME : spécifiez le nom du cluster Dataproc Flink auquel envoyer la tâche.
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 tâches Flink sont répertoriées sur la page Page Jobs de Dataproc dans la console Google Cloud.
- Vous pouvez cliquer sur Arrêter ou Supprimer sur la page Tâches ou Informations sur la tâche. dans la console Google Cloud pour arrêter ou supprimer un job.
Exécuter des tâches Flink à l'aide de la CLI flink
Au lieu de
Exécuter des jobs Flink à l'aide de la ressource Dataproc Jobs
vous pouvez exécuter des jobs 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 une tâche de CLI flink
sur
votre cluster Dataproc Flink.
Connectez-vous en SSH au nœud maître:utilisez la méthode SSH pour ouvrir une fenêtre de terminal sur la VM maître du cluster.
Définissez le chemin d'accès des classes:initialisez le chemin de classe Hadoop à partir de la fenêtre du terminal SSH sur le VM maître du cluster Flink:
export HADOOP_CLASSPATH=$(hadoop classpath)
Exécuter des tâches Flink : vous pouvez exécuter des tâches Flink dans différents modes de déploiement sur YARN : application, par tâche et session.
Mode application:le mode Application Flink est compatible avec les images Dataproc version 2.0 et ultérieures. Ce mode exécute la méthode
main()
de la tâche sur le gestionnaire de tâches YARN. Le cluster s'arrête une fois le job terminé.Exemple d'envoi d'une 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 tâches en cours d'exécution :
./bin/flink list -t yarn-application -Dyarn.application.id=application_XXXX_YY
Annulez une tâche en cours d'exécution:
./bin/flink cancel -t yarn-application -Dyarn.application.id=application_XXXX_YY <jobId>
Mode par tâche : ce mode Flink exécute la méthode
main()
de la tâche côté client.Exemple d'envoi d'une tâche :
flink run \ -m yarn-cluster \ -p 4 \ -ys 2 \ -yjm 1024m \ -ytm 2048m \ /usr/lib/flink/examples/batch/WordCount.jar
Mode session:démarrez une session Flink YARN de longue durée, puis envoyez une ou plusieurs tâches à la session.
Démarrer une session:vous pouvez démarrer une session Flink dans l'un des de différentes manières:
Créez un cluster Flink en ajoutant le
--metadata flink-start-yarn-session=true
au paramètregcloud dataproc clusters create
(voir 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 application YARN de la session est enregistré dans
/tmp/.yarn-properties-${USER}
. Vous pouvez lister l'ID à l'aide de la commandeyarn application -list
.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
Exécutez le script wrapper Flink de
/usr/bin/flink-yarn-daemon
avec paramètres par défaut:. /usr/bin/flink-yarn-daemon
Envoyer un job à une session:exécutez la commande suivante pour envoyer un Appliquez un clin d'œil à 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.
Supprimer
http:// prefix
de l'URL. Cette URL est listée dans la sortie de la commande lorsque vous démarrez une session Flink. Vous pouvez exécuter la commande suivante pour répertorier cette URL dans le champTracking-URL
:
yarn application -list -appId=<yarn-app-id> | sed 's#http://##' ```
- 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.
Supprimer
Lister les tâches dans une session : pour lister les tâches Flink dans une session, procédez comme suit :
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 d'application YARN de la session à partir de
/tmp/.yarn-properties-${USER}
ou de la sortie deyarn application -list
, puis exécutez<code>
flink list -yid YARN_APPLICATION_ID.Exécutez
flink list -m FLINK_MASTER_URL
.
Stop a session (Arrêter une session) : pour arrêter la session, obtenez l'ID application YARN. de la session de
/tmp/.yarn-properties-${USER}
ou la sortie deyarn 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 tâches 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:
- Tâches Java Beam
- 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.
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
: permet à votre cluster d'accéder aux API Google Cloud. (voir Bonnes pratiques concernant les niveaux d'accès). Le niveau d'accès "cloud-platform
" est activé par défaut (vous n'avez pas besoin d'inclure ce paramètre) lorsque vous créez un cluster qui utilise la version d'image 2.1 ou ultérieure de Dataproc.
Utilisez l'utilitaire SSH pour ouvrir une fenêtre de terminal sur le nœud maître du cluster Flink.
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
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. Voir la page Compatibilité des versions Flink qui liste la compatibilité des versions 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
Empaqueter l'exemple de décompte de mots.
mvn package -Pflink-runner
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 utilisergcloud storage cp
pour accélérer les transferts de fichiers vers votre cluster Dataproc à partir de Cloud Storage.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/
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 .
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
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
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 dans d'autres langages compatibles, vous pouvez :
utilisez FlinkRunner
et PortableRunner
comme décrit sur le schéma
Flink Runner
(consultez également la page 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.
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
: 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
: active l'accès à l'interface utilisateur de Flink Job Manager.--scopes
: permet à votre cluster d'accéder aux API Google Cloud (voir Bonnes pratiques concernant les niveaux d'accès). Le champ d'applicationcloud-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.
Utiliser la gcloud CLI en local ou dans Cloud Shell pour créer bucket Cloud Storage. Vous devez spécifier l'élément BUCKET_NAME. lorsque vous exécutez un exemple de programme de décompte de mots.
gcloud storage buckets create BUCKET_NAME
Dans une fenêtre de terminal de 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 tâches sont exécutées. 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 Dataproc. Vous spécifierez FLINK_VERSION lorsque vous exécuter un exemple de programme de décompte de mots.
flink --version
Installez les bibliothèques Python nécessaires à ce job sur le nœud maître du cluster.
Installez une version de Beam compatible avec la version de Flink sur le cluster.
python -m pip install apache-beam[gcp]==BEAM_VERSION
Exécuter l'exemple de nombre de mots sur le maître de cluster d'un nœud.
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, comme indiqué précédemment.--flink_submit_uber_jar
: utilisez l'uber JAR pour exécuter le job Beam.--output
: BUCKET_NAME, créé précédemment.
Vérifiez que les résultats ont bien été écrits dans votre bucket.
gcloud storage cat gs://BUCKET_NAME/python-wordcount-out-SHARD_ID
Arrêtez la session YARN Flink.
- Permet d'obtenir 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, un ticket 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:
- Créez un cluster Dataproc Flink.
- Une fois le cluster créé, cliquez sur la passerelle des composants. Lien ResourceManager YAML dans l'onglet "Interface Web" de la page Détails du cluster dans la console Google Cloud.
- Dans l'interface utilisateur YARN Resource Manager, identifiez l'entrée de l'application de cluster Flink. Selon l'état d'avancement d'une tâche, un objet ApplicationMaster ou Historique s'affiche.
- 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.