Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra. # Commencez avec Amazon Managed Service pour Apache Flink pour Python Cette section présente les concepts fondamentaux d’un service géré pour Apache Flink à l’aide de Python et de l’API de table. Elle décrit les options disponibles pour créer et tester vos applications. Elle fournit également des instructions pour installer les outils nécessaires pour suivre les didacticiels de ce guide et pour créer votre première application. **Topics** + [Passez en revue les composants d'un service géré pour une application Apache Flink](#gs-python-table-components) + [Remplir les conditions préalables](#gs-python-prerequisites) + [Création et exécution d'un service géré pour l'application Apache Flink pour Python](gs-python-createapp.md) + [Nettoyer les AWS ressources](gs-python-cleanup.md) ## Passez en revue les composants d'un service géré pour une application Apache Flink **Note** Amazon Managed Service pour Apache Flink prend en charge tous les [Apache APIs Flink](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/concepts/overview/#flinks-apis). Selon l'API que vous choisissez, la structure de l'application est légèrement différente. Une approche courante lors du développement d'une application Apache Flink en Python consiste à définir le flux de l'application à l'aide du code SQL intégré au code Python. C'est l'approche que nous suivons dans le didacticiel Gettgin Started suivant. Pour traiter les données, votre application Managed Service for Apache Flink utilise un script Python pour définir le flux de données qui traite les entrées et produit les sorties à l'aide du moteur d'exécution Apache Flink. Un service géré typique pour une application Apache Flink comprend les composants suivants : + **Propriétés d’exécution :** vous pouvez utiliser les *propriétés d’exécution* pour configurer votre application sans recompiler le code de votre application. + **Sources :** l'application consomme des données provenant d'une ou de plusieurs *sources*. Une source utilise un [connecteur](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/connectors/table/overview/) pour lire les données d'un système externe tel qu'un flux de données Kinesis ou une rubrique Amazon MSK. Vous pouvez également utiliser des connecteurs spéciaux pour générer des données depuis l'application. Lorsque vous utilisez SQL, l'application définit les sources sous forme de *tables sources*. + **Transformations :** l'application traite les données en utilisant une ou plusieurs *transformations* qui peuvent filtrer, enrichir ou agréger les données. Lorsque vous utilisez SQL, l'application définit les transformations sous forme de requêtes SQL. + **Récepteurs :** l'application envoie des données à des sources externes par le biais *de récepteurs*. Un récepteur utilise un [connecteur](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/connectors/table/overview/) pour envoyer des données vers un système externe tel qu'un flux de données Kinesis, une rubrique Amazon MSK, un compartiment Amazon S3 ou une base de données relationnelle. Vous pouvez également utiliser un connecteur spécial pour imprimer la sortie à des fins de développement. Lorsque vous utilisez SQL, l'application définit les récepteurs comme des *tables réceptrices* dans lesquelles vous insérez les résultats. Pour de plus amples informations, veuillez consulter [Écrire des données à l'aide de récepteurs dans le service géré pour Apache Flink](how-sinks.md). Votre application Python peut également nécessiter des dépendances externes, telles que des bibliothèques Python supplémentaires ou tout connecteur Flink utilisé par votre application. Lorsque vous créez un package pour votre application, vous devez inclure toutes les dépendances dont elle a besoin. Ce didacticiel explique comment inclure les dépendances des connecteurs et comment empaqueter l'application en vue de son déploiement sur Amazon Managed Service pour Apache Flink. ## Remplir les conditions préalables Pour terminer ce didacticiel, vous devez disposer des éléments suivants : + **Python 3.11**[, utilisant de préférence un environnement autonome tel que [VirtualEnv (venv)](https://docs.python.org/3.11/library/venv.html), [Conda ou Miniconda](https://docs.conda.io/en/latest/).](https://docs.anaconda.com/miniconda/) + [Client Git](https://git-scm.com/book/en/v2/Getting-Started-Installing-Git) : installez le client Git si ce n'est pas déjà fait. + [Kit de développement Java (JDK) version 11](https://www.oracle.com/java/technologies/downloads/#java11) : installez un JDK Java 11 et définissez la variable d'`JAVA_HOME`environnement pour qu'elle pointe vers l'emplacement de votre installation. Si vous n'avez pas de JDK 11, vous pouvez utiliser [Amazon Corretto](https://docs.aws.amazon.com/corretto)n'importe quel JDK standard de notre choix. + Pour vérifier que le JDK est correctement installé, exécutez la commande suivante. Le résultat sera différent si vous utilisez un JDK autre qu'Amazon Corretto 11. Assurez-vous que la version est 11.x. ``` $ java --version openjdk 11.0.23 2024-04-16 LTS OpenJDK Runtime Environment Corretto-11.0.23.9.1 (build 11.0.23+9-LTS) OpenJDK 64-Bit Server VM Corretto-11.0.23.9.1 (build 11.0.23+9-LTS, mixed mode) ``` + [Apache Maven](https://maven.apache.org/) : installez Apache Maven si ce n'est pas déjà fait. Pour plus d'informations, consultez la section [Installation d'Apache Maven](https://maven.apache.org/install.html). + Pour tester votre installation d'Apache Maven, utilisez la commande suivante : ``` $ mvn -version ``` **Note** Bien que votre application soit écrite en Python, Apache Flink s'exécute dans la machine virtuelle Java (JVM). Il distribue la plupart des dépendances, telles que le connecteur Kinesis, sous forme de fichiers JAR. Pour gérer ces dépendances et pour empaqueter l'application dans un fichier ZIP, utilisez [Apache Maven](https://maven.apache.org/). Ce didacticiel explique comment procéder. **Avertissement** Nous vous recommandons d'utiliser Python 3.11 pour le développement local. Il s'agit de la même version de Python utilisée par Amazon Managed Service pour Apache Flink avec le moteur d'exécution Flink 1.19. L'installation de la bibliothèque Python Flink 1.19 sur Python 3.12 peut échouer. Si une autre version de Python est installée par défaut sur votre machine, nous vous recommandons de créer un environnement autonome tel que VirtualEnv Python 3.11. **IDE pour le développement local** Nous vous recommandons d'utiliser un environnement de développement tel que [PyCharm](https://www.jetbrains.com/pycharm/)[Visual Studio Code](https://code.visualstudio.com/) pour développer et compiler votre application. Effectuez ensuite les deux premières étapes du [Commencez avec Amazon Managed Service pour Apache Flink (DataStream API)](getting-started.md) : + [Configuration d'un AWS compte et création d'un utilisateur administrateur](setting-up.md) + [Configurez le AWS Command Line Interface (AWS CLI)](setup-awscli.md) Consultez [Création d’une application](gs-python-createapp.md) pour démarrer. # Création et exécution d'un service géré pour l'application Apache Flink pour Python Dans cette section, vous allez créer un service géré pour une application Apache Flink pour Python avec un flux Kinesis comme source et comme récepteur. **Topics** + [Création de ressources dépendantes](#gs-python-resources) + [Configuration de votre environnement de développement local](#gs-python-set-up) + [Téléchargez et examinez le code Python de streaming d'Apache Flink](#gs-python-download) + [Gérer les dépendances JAR](#gs-python-jar-dependencies) + [Écrire des exemples d'enregistrements dans le flux d'entrée](#gs-python-sample-records) + [Exécutez votre application localement](#gs-python-run-locally) + [Observez les données d'entrée et de sortie dans les flux Kinesis](#gs-python-observe-input-output) + [Arrêtez l'exécution locale de votre application](#gs-python-stop) + [Package du code de votre application](#gs-python-package-code) + [Téléchargez le package d'application dans un compartiment Amazon S3](#gs-python-upload-bucket) + [Création et configuration du service géré pour l'application Apache Flink](#gs-python-7) + [Étape suivante](#gs-python-next-step-4) ## Création de ressources dépendantes Avant de créer un service géré pour Apache Flink dans le cadre de cet exercice, vous commencez par créer les ressources dépendantes suivantes : + Deux flux Kinesis pour l’entrée et la sortie. + Un compartiment Amazon S3 pour stocker le code de l'application. **Note** Ce didacticiel part du principe que vous déployez votre application dans la région us-east-1. Si vous utilisez une autre région, vous devez adapter toutes les étapes en conséquence. ### Création de deux flux Kinesis Avant de créer une application Managed Service for Apache Flink pour cet exercice, créez deux flux de données Kinesis `ExampleInputStream` (`ExampleOutputStream`et) dans la même région que vous utiliserez pour déployer votre application (us-east-1 dans cet exemple). Votre application utilise ces flux pour les flux source et de destination de l’application. Vous pouvez créer ces flux à l’aide de la console Amazon Kinesis ou de la commande AWS CLI suivante. Pour obtenir des instructions sur la console, consultez [Création et mise à jour de flux de données](https://docs.aws.amazon.com/kinesis/latest/dev/amazon-kinesis-streams.html) dans le *Guide du développeur Amazon Kinesis Data Streams*. **Pour créer les flux de données (AWS CLI)** 1. Pour créer le premier flux (`ExampleInputStream`), utilisez la commande Amazon Kinesis `create-stream` AWS CLI suivante. ``` $ aws kinesis create-stream \ --stream-name ExampleInputStream \ --shard-count 1 \ --region us-east-1 ``` 1. Pour créer le second flux utilisé par l’application pour écrire la sortie, exécutez la même commande en remplaçant le nom du flux par `ExampleOutputStream`. ``` $ aws kinesis create-stream \ --stream-name ExampleOutputStream \ --shard-count 1 \ --region us-east-1 ``` ### Créer un compartiment Amazon S3 Vous pouvez créer un compartiment Amazon S3 à l’aide de la console. Pour obtenir les instructions relatives à la création de cette ressource, consultez les rubriques suivantes : + [Comment créer un compartiment S3 ?](https://docs.aws.amazon.com/AmazonS3/latest/userguide/create-bucket.html) dans le *Guide de l’utilisateur de la console Amazon Simple Storage Service*. Donnez au compartiment Amazon S3 un nom unique au monde, par exemple en ajoutant votre nom de connexion. **Note** Assurez-vous de créer le compartiment S3 dans la région que vous utilisez pour ce didacticiel (us-east-1). ### Autres ressources Lorsque vous créez votre application, Managed Service for Apache Flink crée les CloudWatch ressources Amazon suivantes si elles n'existent pas déjà : + Un groupe de journaux appelé `/AWS/KinesisAnalytics-java/`. + Un flux de journaux appelé `kinesis-analytics-log-stream` ## Configuration de votre environnement de développement local Pour le développement et le débogage, vous pouvez exécuter l'application Python Flink sur votre machine. Vous pouvez démarrer l'application depuis la ligne de commande avec `python main.py` ou dans l'IDE Python de votre choix. **Note** Python 3.10 ou 3.11, Java 11, Apache Maven et Git doivent être installés sur votre machine de développement. Nous vous recommandons d'utiliser un IDE tel que [PyCharm](https://www.jetbrains.com/pycharm/)[Visual Studio Code](https://code.visualstudio.com/). Pour vérifier que vous répondez à tous les prérequis, reportez-vous à la section [Remplir les conditions préalables pour terminer les exercices](gs-python.md#gs-python-prerequisites) Avant de continuer. ### Installation de la PyFlink bibliothèque Pour développer votre application et l'exécuter localement, vous devez installer la bibliothèque Python Flink. 1. Créez un environnement Python autonome à l'aide VirtualEnv de Conda ou de tout autre outil Python similaire. 1. Installez la PyFlink bibliothèque dans cet environnement. Utilisez la même version d'exécution d'Apache Flink que celle que vous utiliserez dans Amazon Managed Service pour Apache Flink. Actuellement, le temps d'exécution recommandé est 1.19.1. ``` $ pip install apache-flink==1.19.1 ``` 1. Assurez-vous que l'environnement est actif lorsque vous exécutez votre application. Si vous exécutez l'application dans l'IDE, assurez-vous que l'IDE utilise l'environnement comme environnement d'exécution. Le processus dépend de l'IDE que vous utilisez. **Note** Il vous suffit d'installer la PyFlink bibliothèque. Il **n'est pas** nécessaire d'installer un cluster Apache Flink sur votre machine. ### Authentifiez votre session AWS L'application utilise les flux de données Kinesis pour publier des données. Lors de l'exécution locale, vous devez disposer d'une session AWS authentifiée valide avec les autorisations nécessaires pour écrire dans le flux de données Kinesis. Procédez comme suit pour authentifier votre session : 1. Si vous n'avez pas configuré le profil AWS CLI et un profil nommé avec des informations d'identification valides, consultez[Configurez le AWS Command Line Interface (AWS CLI)](setup-awscli.md). 1. Vérifiez que vous êtes correctement AWS CLI configuré et que vos utilisateurs sont autorisés à écrire dans le flux de données Kinesis en publiant l'enregistrement de test suivant : ``` $ aws kinesis put-record --stream-name ExampleOutputStream --data TEST --partition-key TEST ``` 1. Si votre IDE dispose d'un plugin à intégrer AWS, vous pouvez l'utiliser pour transmettre les informations d'identification à l'application exécutée dans l'IDE. Pour plus d'informations, consultez [AWS Toolkit for PyCharm](https://aws.amazon.com/pycharm/), [AWS Toolkit for Visual Studio Code](https://aws.amazon.com/visualstudiocode/) et [AWS Toolkit for IntelliJ IDEA](https://aws.amazon.com/intellij/). ## Téléchargez et examinez le code Python de streaming d'Apache Flink Le code de l'application Python pour cet exemple est disponible sur GitHub. Pour télécharger le code d’application, procédez comme suit : 1. Cloner le référentiel distant à l’aide de la commande suivante : ``` git clone https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples.git ``` 1. Accédez au répertoire `./python/GettingStarted`. ### Vérifiez les composants de l'application Le code de l'application se trouve dans`main.py`. Nous utilisons le SQL intégré à Python pour définir le flux de l'application. **Note** Pour une expérience de développement optimisée, l'application est conçue pour s'exécuter sans aucune modification de code à la fois sur Amazon Managed Service pour Apache Flink et localement, pour le développement sur votre machine. L'application utilise la variable `IS_LOCAL = true` d'environnement pour détecter si elle est exécutée localement. Vous devez définir la variable `IS_LOCAL = true` d'environnement sur votre shell ou dans la configuration d'exécution de votre IDE. + L'application configure l'environnement d'exécution et lit la configuration d'exécution. Pour fonctionner à la fois sur Amazon Managed Service pour Apache Flink et localement, l'application vérifie la `IS_LOCAL` variable. + Voici le comportement par défaut lorsque l'application s'exécute dans Amazon Managed Service pour Apache Flink : 1. Chargez les dépendances fournies avec l'application. Pour plus d'informations, voir (lien) 1. Chargez la configuration à partir des propriétés d'exécution que vous définissez dans l'application Amazon Managed Service for Apache Flink. Pour plus d'informations, voir (lien) + Lorsque l'application détecte `IS_LOCAL = true` que vous l'exécutez localement : 1. Charge les dépendances externes depuis le projet. 1. Charge la configuration à partir du `application_properties.json` fichier inclus dans le projet. ``` ... APPLICATION_PROPERTIES_FILE_PATH = "/etc/flink/application_properties.json" ... is_local = ( True if os.environ.get("IS_LOCAL") else False ) ... if is_local: APPLICATION_PROPERTIES_FILE_PATH = "application_properties.json" CURRENT_DIR = os.path.dirname(os.path.realpath(__file__)) table_env.get_config().get_configuration().set_string( "pipeline.jars", "file:///" + CURRENT_DIR + "/target/pyflink-dependencies.jar", ) ``` + L'application définit une table source avec une `CREATE TABLE` instruction, à l'aide du connecteur [Kinesis](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/connectors/table/kinesis/). Ce tableau lit les données du flux Kinesis en entrée. L'application prend le nom du flux, la région et la position initiale de la configuration d'exécution. ``` table_env.execute_sql(f""" CREATE TABLE prices ( ticker VARCHAR(6), price DOUBLE, event_time TIMESTAMP(3), WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND ) PARTITIONED BY (ticker) WITH ( 'connector' = 'kinesis', 'stream' = '{input_stream_name}', 'aws.region' = '{input_stream_region}', 'format' = 'json', 'json.timestamp-format.standard' = 'ISO-8601' ) """) ``` + Dans cet exemple, l'application définit également une table réceptrice à l'aide [du connecteur Kinesis](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/connectors/table/kinesis/). Ce conte envoie des données au flux Kinesis de sortie. ``` table_env.execute_sql(f""" CREATE TABLE output ( ticker VARCHAR(6), price DOUBLE, event_time TIMESTAMP(3) ) PARTITIONED BY (ticker) WITH ( 'connector' = 'kinesis', 'stream' = '{output_stream_name}', 'aws.region' = '{output_stream_region}', 'sink.partitioner-field-delimiter' = ';', 'sink.batch.max-size' = '100', 'format' = 'json', 'json.timestamp-format.standard' = 'ISO-8601' )""") ``` + Enfin, l'application exécute un code SQL que `INSERT INTO...` la table réceptrice contient à partir de la table source. Dans une application plus complexe, vous devez probablement effectuer des étapes supplémentaires pour transformer les données avant de les écrire dans le récepteur. ``` table_result = table_env.execute_sql("""INSERT INTO output SELECT ticker, price, event_time FROM prices""") ``` + Vous devez ajouter une autre étape à la fin de la `main()` fonction pour exécuter l'application localement : ``` if is_local: table_result.wait() ``` Sans cette instruction, l'application s'arrête immédiatement lorsque vous l'exécutez localement. Vous ne devez pas exécuter cette instruction lorsque vous exécutez votre application dans Amazon Managed Service pour Apache Flink. ## Gérer les dépendances JAR Une PyFlink application nécessite généralement un ou plusieurs connecteurs. L'application présentée dans ce didacticiel utilise le [connecteur Kinesis](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/connectors/table/kinesis/). Apache Flink s'exécutant dans la JVM Java, les connecteurs sont distribués sous forme de fichiers JAR, que vous implémentiez ou non votre application en Python. Vous devez intégrer ces dépendances à l'application lorsque vous la déployez sur Amazon Managed Service pour Apache Flink. Dans cet exemple, nous montrons comment utiliser Apache Maven pour récupérer les dépendances et empaqueter l'application à exécuter sur le service géré pour Apache Flink. **Note** Il existe d'autres méthodes pour récupérer et empaqueter les dépendances. Cet exemple illustre une méthode qui fonctionne correctement avec un ou plusieurs connecteurs. Il vous permet également d'exécuter l'application localement, à des fins de développement et sur le service géré pour Apache Flink sans modifier le code. ### Utilisez le fichier pom.xml Apache Maven utilise le `pom.xml` fichier pour contrôler les dépendances et le packaging des applications. Toutes les dépendances JAR sont spécifiées dans le `pom.xml` fichier du `...` bloc. ``` 4.0.0 ... org.apache.flink flink-connector-kinesis 4.3.0-1.19 ... ``` Pour trouver l'artefact et la version du connecteur appropriés à utiliser, voir[Utiliser les connecteurs Apache Flink avec le service géré pour Apache Flink](how-flink-connectors.md). Assurez-vous de vous référer à la version d'Apache Flink que vous utilisez. Pour cet exemple, nous utilisons le connecteur Kinesis. Pour Apache Flink 1.19, la version du connecteur est. `4.3.0-1.19` **Note** Si vous utilisez Apache Flink 1.19, aucune version de connecteur n'a été publiée spécifiquement pour cette version. Utilisez les connecteurs publiés pour la version 1.18. ### Dépendances relatives aux téléchargements et aux packages Utilisez Maven pour télécharger les dépendances définies dans le `pom.xml` fichier et les empaqueter pour l'application Python Flink. 1. Accédez au répertoire qui contient le projet Python Getting Started appelé`python/GettingStarted`. 1. Exécutez la commande suivante : ``` $ mvn package ``` Maven crée un nouveau fichier appelé`./target/pyflink-dependencies.jar`. Lorsque vous développez localement sur votre machine, l'application Python recherche ce fichier. **Note** Si vous oubliez d'exécuter cette commande, lorsque vous essayez d'exécuter votre application, elle échouera avec le message d'erreur suivant : **Impossible de trouver une usine pour l'identifiant « kinesis** ». ## Écrire des exemples d'enregistrements dans le flux d'entrée Dans cette section, vous allez envoyer des exemples d'enregistrements au flux pour que la demande soit traitée. Deux options s'offrent à vous pour générer des exemples de données, soit à l'aide d'un script Python, soit à l'aide du [Kinesis Data](https://github.com/awslabs/amazon-kinesis-data-generator) Generator. ### Générer des exemples de données à l'aide d'un script Python Vous pouvez utiliser un script Python pour envoyer des exemples d'enregistrements au flux. **Note** Pour exécuter ce script Python, vous devez utiliser Python 3.x et installer la bibliothèque du [AWS SDK pour Python (Boto)](https://aws.amazon.com/developer/language/python/). **Pour commencer à envoyer des données de test vers le flux d'entrée Kinesis :** 1. Téléchargez le script `stock.py` Python du générateur de données depuis le [ GitHub référentiel du générateur de données](https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/tree/main/python/data-generator). 1. Exécutez le script `stock.py` : ``` $ python stock.py ``` Continuez à exécuter le script pendant que vous terminez le reste du didacticiel. Vous pouvez désormais exécuter votre application Apache Flink. ### Génération d'échantillons de données à l'aide de Kinesis Data Generator Au lieu d'utiliser le script Python, vous pouvez utiliser [Kinesis Data Generator](https://github.com/awslabs/amazon-kinesis-data-generator), également disponible dans une [version hébergée](https://awslabs.github.io/amazon-kinesis-data-generator/web/producer.html), pour envoyer des échantillons de données aléatoires au flux. Kinesis Data Generator s'exécute dans votre navigateur et vous n'avez rien à installer sur votre machine. **Pour configurer et exécuter Kinesis Data Generator, procédez comme suit :** 1. Suivez les instructions de la [documentation de Kinesis Data Generator](https://awslabs.github.io/amazon-kinesis-data-generator/web/help.html) pour configurer l'accès à l'outil. Vous allez exécuter un CloudFormation modèle qui définit un utilisateur et un mot de passe. 1. Accédez à Kinesis Data Generator via l'URL générée par le CloudFormation modèle. Vous pouvez trouver l'URL dans l'onglet **Sortie** une fois le CloudFormation modèle terminé. 1. Configurez le générateur de données : + **Région :** Sélectionnez la région que vous utilisez pour ce didacticiel : us-east-1 + **Stream/flux de diffusion :** sélectionnez le flux d'entrée que l'application utilisera : `ExampleInputStream` + **Enregistrements par seconde :** 100 + **Modèle d'enregistrement :** copiez et collez le modèle suivant : ``` { "event_time" : "{{date.now("YYYY-MM-DDTkk:mm:ss.SSSSS")}}, "ticker" : "{{random.arrayElement( ["AAPL", "AMZN", "MSFT", "INTC", "TBV"] )}}", "price" : {{random.number(100)}} } ``` 1. Testez le modèle : choisissez le **modèle de test** et vérifiez que l'enregistrement généré est similaire au suivant : ``` { "event_time" : "2024-06-12T15:08:32.04800, "ticker" : "INTC", "price" : 7 } ``` 1. Démarrez le générateur de données : Choisissez **Sélectionner envoyer les données**. Kinesis Data Generator envoie désormais des données au. `ExampleInputStream` ## Exécutez votre application localement Vous pouvez tester l'application localement, en l'exécutant depuis la ligne de commande avec `python main.py` ou depuis votre IDE. Pour exécuter votre application localement, la version correcte de la PyFlink bibliothèque doit être installée, comme décrit dans la section précédente. Pour plus d'informations, voir (lien) **Note** Avant de continuer, vérifiez que les flux d'entrée et de sortie sont disponibles. Consultez [Création de deux flux de données Amazon Kinesis](get-started-exercise.md#get-started-exercise-1). Vérifiez également que vous êtes autorisé à lire et à écrire à partir des deux flux. Consultez [Authentifiez votre session AWS](get-started-exercise.md#get-started-exercise-2-5). ### Importez le projet Python dans votre IDE Pour commencer à travailler sur l'application dans votre IDE, vous devez l'importer en tant que projet Python. Le référentiel que vous avez cloné contient plusieurs exemples. Chaque exemple est un projet distinct. Pour ce didacticiel, importez le contenu du `./python/GettingStarted` sous-répertoire dans votre IDE. Importez le code en tant que projet Python existant. **Note** Le processus exact d'importation d'un nouveau projet Python varie en fonction de l'IDE que vous utilisez. ### Vérifiez la configuration de l'application locale Lorsqu'elle est exécutée localement, l'application utilise la configuration contenue dans le `application_properties.json` fichier situé dans le dossier des ressources du projet situé sous`./src/main/resources`. Vous pouvez modifier ce fichier pour utiliser différents noms de flux Kinesis ou différentes régions. ``` [ { "PropertyGroupId": "InputStream0", "PropertyMap": { "stream.name": "ExampleInputStream", "flink.stream.initpos": "LATEST", "aws.region": "us-east-1" } }, { "PropertyGroupId": "OutputStream0", "PropertyMap": { "stream.name": "ExampleOutputStream", "aws.region": "us-east-1" } } ] ``` ### Exécutez votre application Python localement Vous pouvez exécuter votre application localement, soit depuis la ligne de commande en tant que script Python normal, soit depuis l'IDE. **Pour exécuter votre application à partir de la ligne de commande** 1. Assurez-vous que l'environnement Python autonome tel que Conda ou celui dans VirtualEnv lequel vous avez installé la bibliothèque Python Flink est actuellement actif. 1. Assurez-vous d'avoir couru `mvn package` au moins une fois. 1. Définissez la variable d'environnement :`IS_LOCAL = true` ``` $ export IS_LOCAL=true ``` 1. Exécutez l'application sous la forme d'un script Python normal. ``` $python main.py ``` **Pour exécuter l'application depuis l'IDE** 1. Configurez votre IDE pour exécuter le `main.py` script avec la configuration suivante : 1. Utilisez l'environnement Python autonome tel que Conda ou celui dans VirtualEnv lequel vous avez installé la PyFlink bibliothèque. 1. Utilisez les AWS informations d'identification pour accéder aux flux de données Kinesis en entrée et en sortie. 1. Configurez `IS_LOCAL = true`. 1. Le processus exact pour définir la configuration d'exécution dépend de votre IDE et varie. 1. Lorsque vous avez configuré votre IDE, exécutez le script Python et utilisez les outils fournis par votre IDE pendant que l'application est en cours d'exécution. ### Inspectez les journaux des applications localement Lorsqu'elle est exécutée localement, l'application n'affiche aucun journal dans la console, à l'exception de quelques lignes imprimées et affichées au démarrage de l'application. PyFlink écrit les journaux dans un fichier du répertoire où la bibliothèque Python Flink est installée. L'application imprime l'emplacement des journaux au démarrage. Vous pouvez également exécuter la commande suivante pour trouver les journaux : ``` $ python -c "import pyflink;import os;print(os.path.dirname(os.path.abspath(pyflink.__file__))+'/log')" ``` 1. Répertoriez les fichiers dans le répertoire de journalisation. Vous ne trouvez généralement qu'un seul `.log` fichier. 1. Suivez le fichier pendant que l'application est en cours d'exécution :`tail -f /.log`. ## Observez les données d'entrée et de sortie dans les flux Kinesis Vous pouvez observer les enregistrements envoyés au flux d'entrée par le (générateur d'un exemple de Python) ou le générateur de données Kinesis (lien) à l'aide du **visualiseur de données** de la console Amazon Kinesis. **Pour observer les enregistrements :** [Ouvrez la console Kinesis à l'adresse /kinesis. https://console.aws.amazon.com](https://console.aws.amazon.com/kinesis) Vérifiez que la région est la même que celle dans laquelle vous exécutez ce didacticiel, à savoir us-east-1 USA East (Virginie du Nord) par défaut. Modifiez la région si elle ne correspond pas. Choisissez **Data Streams**. Sélectionnez le flux que vous souhaitez observer, `ExampleInputStream` soit `ExampleOutputStream.` Choisissez l'onglet **Visionneuse de données**. Choisissez n'importe quel **Shard**, conservez **Latest** comme **position de départ**, puis choisissez **Get records**. Le message d'erreur « Aucun enregistrement trouvé pour cette demande » peut s'afficher. Si tel est le cas, choisissez **Réessayer d'obtenir des enregistrements**. Les derniers enregistrements publiés sur le stream s'affichent. Choisissez la valeur dans la colonne Données pour inspecter le contenu de l'enregistrement au format JSON. ## Arrêtez l'exécution locale de votre application Arrêtez l'exécution de l'application dans votre IDE. L'IDE fournit généralement une option « stop ». L'emplacement exact et la méthode dépendent de l'IDE. ## Package du code de votre application Dans cette section, vous allez utiliser Apache Maven pour empaqueter le code de l'application et toutes les dépendances requises dans un fichier .zip. Exécutez à nouveau la commande du package Maven : ``` $ mvn package ``` Cette commande génère le fichier`target/managed-flink-pyflink-getting-started-1.0.0.zip`. ## Téléchargez le package d'application dans un compartiment Amazon S3 Dans cette section, vous allez charger le fichier .zip que vous avez créé dans la section précédente dans le bucket Amazon Simple Storage Service (Amazon S3) que vous avez créé au début de ce didacticiel. Si vous n'avez pas terminé cette étape, consultez (lien). **Pour télécharger le fichier JAR du code de l'application** 1. Ouvrez la console Amazon S3 à l'adresse [https://console.aws.amazon.com/s3/](https://console.aws.amazon.com/s3/). 1. Choisissez le bucket que vous avez créé précédemment pour le code de l'application. 1. Choisissez **Charger**. 1. Choisissez **Add files**. 1. Accédez au fichier .zip généré à l'étape précédente :`target/managed-flink-pyflink-getting-started-1.0.0.zip`. 1. Choisissez **Télécharger** sans modifier les autres paramètres. ## Création et configuration du service géré pour l'application Apache Flink Vous pouvez créer et configurer un service géré pour l'application Apache Flink à l'aide de la console ou du AWS CLI. Pour ce tutoriel, nous allons utiliser la console. ### Pour créer l’application 1. Connectez-vous à la AWS Management Console console Amazon MSF et ouvrez-la à https://console.aws.amazon.com l'adresse /flink. 1. Vérifiez que la bonne région est sélectionnée : USA Est (Virginie du Nord) us-east-1. 1. Ouvrez le menu de droite et choisissez **Applications Apache Flink**, puis **Créer une application de streaming**. Vous pouvez également choisir **Créer une application de streaming** dans la section **Commencer** de la page initiale. 1. Sur la page **Créer des applications de streaming** : + Pour **Choisir une méthode pour configurer l'application de traitement des flux**, choisissez **Créer à partir de zéro**. + Pour la **configuration d'Apache Flink, version Application Flink**, choisissez **Apache Flink** 1.19. + Pour la **configuration de l'application** : + Pour **Nom de l’application**, saisissez **MyApplication**. + Pour **Description**, saisissez **My Python test app**. + Dans **Accès aux ressources de l'application**, choisissez **Create/update IAM role kinesis-analytics-MyApplication-us -east-1** avec les politiques requises. + Pour les **paramètres du modèle pour les applications** : + Dans **Modèles**, sélectionnez **Développement**. + Choisissez **Créer une application de streaming**. **Note** Lorsque vous créez une application de service géré pour Apache Flink à l’aide de la console, vous avez la possibilité de créer un rôle et une politique IAM pour votre application. Votre application utilise ce rôle et cette politique pour accéder à ses ressources dépendantes. Ces ressources IAM sont nommées en utilisant le nom de votre application et la région, comme suit : Stratégie : `kinesis-analytics-service-MyApplication-us-west-2` Rôle : `kinesisanalytics-MyApplication-us-west-2` Amazon Managed Service pour Apache Flink était auparavant connu sous le nom de *Kinesis Data Analytics*. Le nom des ressources générées automatiquement est préfixé par un préfixe `kinesis-analytics` pour des raisons de rétrocompatibilité. ### Modifier la politique IAM Modifiez la politique IAM pour ajouter des autorisations afin d’accéder au compartiment Amazon S3. **Pour modifier la politique IAM afin d’ajouter des autorisations au compartiment S3** 1. Ouvrez la console IAM à l’adresse [https://console.aws.amazon.com/iam/](https://console.aws.amazon.com/iam/). 1. Choisissez **Stratégies**. Choisissez la politique **`kinesis-analytics-service-MyApplication-us-east-1`** créée pour vous par la console dans la section précédente. 1. Choisissez **Modifier**, puis sélectionnez l'onglet **JSON**. 1. Ajoutez la section mise en surbrillance dans l’exemple de stratégie suivant à la politique. Remplacez le compte d'exemple IDs (*012345678901*) par votre identifiant de compte. ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "ReadCode", "Effect": "Allow", "Action": [ "s3:GetObject", "s3:GetObjectVersion" ], "Resource": [ "arn:aws:s3:::my-bucket/kinesis-analytics-placeholder-s3-object" ] }, { "Sid": "ListCloudwatchLogGroups", "Effect": "Allow", "Action": [ "logs:DescribeLogGroups" ], "Resource": [ "arn:aws:logs:us-east-1:012345678901:log-group:*" ] }, { "Sid": "ListCloudwatchLogStreams", "Effect": "Allow", "Action": [ "logs:DescribeLogStreams" ], "Resource": [ "arn:aws:logs:us-east-1:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:*" ] }, { "Sid": "PutCloudwatchLogs", "Effect": "Allow", "Action": [ "logs:PutLogEvents" ], "Resource": [ "arn:aws:logs:us-east-1:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:kinesis-analytics-log-stream" ] }, { "Sid": "ReadInputStream", "Effect": "Allow", "Action": "kinesis:*", "Resource": "arn:aws:kinesis:us-east-1:012345678901:stream/ExampleInputStream" }, { "Sid": "WriteOutputStream", "Effect": "Allow", "Action": "kinesis:*", "Resource": "arn:aws:kinesis:us-east-1:012345678901:stream/ExampleOutputStream" } ] } ``` ------ 1. Choisissez **Suivant**, puis **Enregistrer les modifications**. ### Configuration de l'application Modifiez la configuration de l'application pour définir l'artefact du code de l'application. **Pour configurer l’application** 1. Sur la **MyApplication**page, choisissez **Configurer**. 1. Dans la section **Emplacement du code de l'application** : + Pour le compartiment **Amazon S3**, sélectionnez le compartiment que vous avez créé précédemment pour le code de l'application. Choisissez **Parcourir** et sélectionnez le compartiment approprié, puis choisissez **Choisir**. Ne sélectionnez pas le nom du bucket. + Pour le **chemin de l'objet Amazon S3**, saisissez **managed-flink-pyflink-getting-started-1.0.0.zip**. 1. Pour les **autorisations d'accès**, choisissez **Créer/mettre à jour le rôle IAM `kinesis-analytics-MyApplication-us-east-1` avec les politiques requises**. 1. Accédez aux **propriétés d'exécution** et conservez les valeurs par défaut pour tous les autres paramètres. 1. Choisissez **Ajouter un nouvel article** et ajoutez chacun des paramètres suivants : [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/fr_fr/managed-flink/latest/java/gs-python-createapp.html) 1. Ne modifiez aucune des autres sections et choisissez **Enregistrer les modifications**. **Note** Lorsque vous choisissez d'activer la CloudWatch journalisation Amazon, Managed Service for Apache Flink crée un groupe de journaux et un flux de journaux pour vous. Les noms de ces ressources sont les suivants : Groupe de journaux : `/aws/kinesis-analytics/MyApplication` Flux de journaux : `kinesis-analytics-log-stream` ### Exécutez l'application L'application est maintenant configurée et prête à être exécutée. **Pour exécuter l’application** 1. Sur la console d'Amazon Managed Service pour Apache Flink, choisissez **My Application**, puis **Run**. 1. Sur la page suivante, page de configuration de la restauration de l'application, choisissez **Exécuter avec le dernier instantané**, puis sélectionnez **Exécuter**. Le **statut** dans **l'application détaille** les transitions entre `Ready` le `Starting` et le `Running` moment où l'application a démarré. Lorsque le `Running` statut de l'application est atteint, vous pouvez désormais ouvrir le tableau de bord Flink. **Pour ouvrir le tableau de bord d'** 1. Choisissez **Ouvrir le tableau de bord Apache Flink**. Le tableau de bord s'ouvre sur une nouvelle page. 1. Dans la **liste des tâches en cours**, choisissez la tâche unique que vous pouvez voir. **Note** Si vous définissez les propriétés d'exécution ou si vous modifiez les politiques IAM de manière incorrecte, le statut de l'application peut devenir`Running`, mais le tableau de bord Flink indique que le travail redémarre continuellement. Il s'agit d'un scénario d'échec courant si l'application est mal configurée ou n'est pas autorisée à accéder aux ressources externes. Dans ce cas, consultez l'onglet **Exceptions** du tableau de bord Flink pour connaître la cause du problème. ### Observez les métriques de l'application en cours d'exécution Sur la **MyApplication**page, dans la section ** CloudWatch des métriques Amazon**, vous pouvez voir certaines des mesures fondamentales de l'application en cours d'exécution. **Pour consulter les statistiques** 1. À côté du bouton **Actualiser**, sélectionnez **10 secondes** dans la liste déroulante. 1. Lorsque l'application est en cours d'exécution et fonctionne correctement, vous pouvez constater une augmentation continue de la métrique de **disponibilité**. 1. La métrique de **redémarrage complet** doit être égale à zéro. S'il augmente, il est possible que la configuration présente des problèmes. Pour étudier le problème, consultez l'onglet **Exceptions** du tableau de bord Flink. 1. La métrique **du nombre de points de contrôle ayant échoué** doit être égale à zéro dans une application saine. **Note** Ce tableau de bord affiche un ensemble fixe de mesures avec une granularité de 5 minutes. Vous pouvez créer un tableau de bord d'application personnalisé avec tous les indicateurs du CloudWatch tableau de bord. ### Observez les données de sortie dans les flux Kinesis Assurez-vous que vous publiez toujours les données en entrée, à l'aide du script Python ou du Kinesis Data Generator. Vous pouvez désormais observer le résultat de l'application exécutée sur le service géré pour Apache Flink en utilisant le visualiseur de données dans le [https://console.aws.amazon.com/kinesis/](https://console.aws.amazon.com/kinesis/), comme vous l'avez déjà fait précédemment. **Pour afficher le résultat** 1. [Ouvrez la console Kinesis à l'adresse /kinesis. https://console.aws.amazon.com](https://console.aws.amazon.com/kinesis) 1. Vérifiez que la région est la même que celle que vous utilisez pour exécuter ce didacticiel. Par défaut, il s'agit de US-East-1US East (Virginie du Nord). Modifiez la région si nécessaire. 1. Choisissez **Data Streams**. 1. Sélectionnez le flux que vous souhaitez observer. Dans le cadre de ce tutoriel, utilisez `ExampleOutputStream`. 1. Choisissez l'onglet **Visionneuse de données**. 1. Sélectionnez n'importe quelle **partition**, conservez **Dernière** comme **position de départ**, puis choisissez **Obtenir des enregistrements**. Le message d'erreur « aucun enregistrement trouvé pour cette demande » peut s'afficher. Si tel est le cas, choisissez **Réessayer d'obtenir des enregistrements**. Les derniers enregistrements publiés sur le stream s'affichent. 1. Sélectionnez la valeur dans la colonne Données pour inspecter le contenu de l'enregistrement au format JSON. ### Arrêtez l'application Pour arrêter l'application, rendez-vous sur la page de console de l'application Managed Service for Apache Flink nommée. `MyApplication` **Pour arrêter l’application** 1. Dans la liste déroulante **Action**, choisissez **Stop**. 1. Le **statut** de **l'application détaille** les transitions entre `Running` le et le `Ready` moment où l'application est complètement arrêtée. `Stopping` **Note** N'oubliez pas d'arrêter également d'envoyer des données au flux d'entrée à partir du script Python ou du Kinesis Data Generator. ## Étape suivante [Nettoyer les AWS ressources](gs-python-cleanup.md) # Nettoyer les AWS ressources Cette section inclut les procédures de nettoyage AWS des ressources créées dans le didacticiel Getting Started (Python). **Topics** + [Supprimer votre application Managed Service for Apache Flink](#gs-python-cleanup-app) + [Supprimer vos flux de données Kinesis](#gs-python-cleanup-msk) + [Supprimer vos objets et votre compartiment Amazon S3](#gs-python-cleanup-s3) + [Supprimer vos ressources IAM](#gs-python-cleanup-iam) + [Supprimer vos CloudWatch ressources](#gs-python-cleanup-cw) ## Supprimer votre application Managed Service for Apache Flink Procédez comme suit pour supprimer l’application. **Pour supprimer l’application** 1. [Ouvrez la console Kinesis à l'adresse /kinesis. https://console.aws.amazon.com](https://console.aws.amazon.com/kinesis) 1. Dans le panneau Managed Service for Apache Flink, choisissez **MyApplication**. 1. Dans la liste déroulante **Actions**, choisissez **Supprimer**, puis confirmez la suppression. ## Supprimer vos flux de données Kinesis 1. Connectez-vous à la AWS Management Console console Amazon MSF et ouvrez-la à https://console.aws.amazon.com l'adresse /flink. 1. Choisissez **Data streams**. 1. Sélectionnez les deux flux que vous avez créés, `ExampleInputStream` et`ExampleOutputStream`. 1. Dans la liste déroulante **Actions**, choisissez **Supprimer**, puis confirmez la suppression. ## Supprimer vos objets et votre compartiment Amazon S3 Procédez comme suit pour supprimer vos objets et votre compartiment S3. **Pour supprimer l'objet du compartiment S3** 1. Ouvrez la console Amazon S3 à l'adresse [https://console.aws.amazon.com/s3/](https://console.aws.amazon.com/s3/). 1. Sélectionnez le compartiment S3 que vous avez créé pour l'artefact d'application. 1. Sélectionnez l'artefact d'application que vous avez chargé, `amazon-msf-java-stream-app-1.0.jar` nommez-le. 1. Choisissez **Supprimer** et confirmez la suppression. **Pour supprimer le compartiment S3** 1. Ouvrez la console Amazon S3 à l'adresse [https://console.aws.amazon.com/s3/](https://console.aws.amazon.com/s3/). 1. Sélectionnez le compartiment que vous avez créé pour les artefacts. 1. Choisissez **Supprimer** et confirmez la suppression. **Note** Le compartiment S3 doit être vide pour pouvoir être supprimé. ## Supprimer vos ressources IAM Utilisez la procédure suivante pour supprimer vos ressources IAM. **Pour supprimer vos ressources IAM** 1. Ouvrez la console IAM à l’adresse [https://console.aws.amazon.com/iam/](https://console.aws.amazon.com/iam/). 1. Dans la barre de navigation, choisissez **Stratégies**. 1. Dans le contrôle du filtre, saisissez **kinesis**. 1. Choisissez la politique **kinesis-analytics-service- MyApplication -us-east-1**. 1. Choisissez **Actions de stratégie**, puis **Supprimer**. 1. Dans la barre de navigation, choisissez **Rôles**. 1. Choisissez le rôle **kinesis-analytics- MyApplication** -us-east-1. 1. Choisissez **Supprimer le rôle**, puis confirmez la suppression. ## Supprimer vos CloudWatch ressources Pour supprimer vos CloudWatch ressources, procédez comme suit. **Pour supprimer vos CloudWatch ressources** 1. Ouvrez la CloudWatch console à l'adresse [https://console.aws.amazon.com/cloudwatch/](https://console.aws.amazon.com/cloudwatch/). 1. Dans la barre de navigation, choisissez **Journaux**. 1. Choisissez le groupe**/aws/kinesis-analytics/MyApplication**log. 1. Choisissez **Supprimer le groupe de journaux**, puis confirmez la suppression.