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. # Chargement de données de streaming dans Amazon OpenSearch Service Vous pouvez utiliser OpenSearch Ingestion pour charger directement [des données de streaming](https://aws.amazon.com/streaming-data/) dans votre domaine Amazon OpenSearch Service, sans avoir besoin de recourir à des solutions tierces. Pour envoyer des données à OpenSearch Ingestion, vous configurez vos producteurs de données et le service fournit automatiquement les données au domaine ou à la collection que vous spécifiez. Pour commencer à utiliser OpenSearch Ingestion, voir[Tutoriel : Ingestion de données dans une collection à l'aide d'Amazon OpenSearch Ingestion](osis-serverless-get-started.md). Vous pouvez toujours utiliser d'autres sources pour charger des données de streaming, telles qu'Amazon Data Firehose et Amazon CloudWatch Logs, qui disposent d'un support intégré pour OpenSearch Service. D'autres, telles qu'Amazon S3, Amazon Kinesis Data Streams, et Amazon DynamoDB, utilisent des fonctions AWS Lambda comme gestionnaires d'événements. Les fonctions Lambda répondent aux nouvelles données en les traitant et en les diffusant dans votre domaine. **Note** Lambda prend en charge différents langages de programmation courants et est disponible dans la plupart des Régions AWS. Pour plus d'informations, consultez [Getting started with Lambda](https://docs.aws.amazon.com/lambda/latest/dg/lambda-app.html) dans le *guide du AWS Lambda développeur* et les [points de terminaison AWS de service](https://docs.aws.amazon.com/general/latest/gr/rande.html#lambda_region) dans le. *Références générales AWS* # Chargement de données de streaming depuis OpenSearch Ingestion Vous pouvez utiliser Amazon OpenSearch Ingestion pour charger des données dans un domaine OpenSearch de service. Vous configurez vos producteurs de données pour qu'ils envoient des données à OpenSearch Ingestion, qui les fournit automatiquement à la collection que vous spécifiez. Vous pouvez également configurer OpenSearch Ingestion pour transformer vos données avant de les livrer. Pour de plus amples informations, veuillez consulter [Présentation d'Amazon OpenSearch Ingestion](ingestion.md). # Chargement de données de streaming à partir d'Amazon S3 Vous pouvez utiliser Lambda pour envoyer des données vers votre domaine de OpenSearch service depuis Amazon S3. Les nouvelles données qui arrivent dans un compartiment S3 déclenchent l'envoi d'une notification d'événement à Lambda, qui exécute alors votre code personnalisé pour effectuer l'indexation. Cette méthode de diffusion de données est extrêmement flexible. Vous pouvez [indexer les métadonnées d'objet](https://aws.amazon.com/blogs/database/indexing-metadata-in-amazon-elasticsearch-service-using-aws-lambda-and-python/), ou si l'objet est en texte brut, analyser et indexer certains éléments du corps de l'objet. Cette section inclut des exemples de code Python simple qui utilisent des expressions régulières pour analyser un fichier journal et indexer les correspondances. ## Conditions préalables Avant de poursuivre, vous devez disposer des ressources suivantes. **** | Prérequis | Description | | --- | --- | | Compartiment Amazon S3 | Pour en savoir plus, consultez [Création de votre premier compartiment S3](https://docs.aws.amazon.com/AmazonS3/latest/userguide/CreatingABucket.html) dans le Guide de l'utilisateur Amazon Simple Storage Service. Le bucket doit résider dans la même région que votre domaine OpenSearch de service. | | OpenSearch Domaine de service | Destination des données après leur traitement par votre fonction Lambda. Pour de plus amples informations, veuillez consulter [Création de domaines OpenSearch de service](createupdatedomains.md#createdomains). | ## Créer le package de déploiement Lambda Les packages de déploiement sont des fichiers ZIP ou JAR qui contiennent votre code et ses dépendances. Cette section inclut des exemples de code Python. Pour les autres langages de programmation, consultez [Packages de déploiement Lambda](https://docs.aws.amazon.com/lambda/latest/dg/gettingstarted-package.html) dans le *Guide du développeur AWS Lambda *. 1. Créez un répertoire. Dans cet exemple, nous utilisons le nom `s3-to-opensearch`. 1. Dans le répertoire, créez un fichier nommé `sample.py` : ``` import boto3 import re import requests from requests_aws4auth import AWS4Auth region = '' # e.g. us-west-1 service = 'es' credentials = boto3.Session().get_credentials() awsauth = AWS4Auth(credentials.access_key, credentials.secret_key, region, service, session_token=credentials.token) host = '' # the OpenSearch Service domain, e.g. https://search-mydomain.us-west-1.es.amazonaws.com index = 'lambda-s3-index' datatype = '_doc' url = host + '/' + index + '/' + datatype headers = { "Content-Type": "application/json" } s3 = boto3.client('s3') # Regular expressions used to parse some simple log lines ip_pattern = re.compile('(\d+\.\d+\.\d+\.\d+)') time_pattern = re.compile('\[(\d+\/\w\w\w\/\d\d\d\d:\d\d:\d\d:\d\d\s-\d\d\d\d)\]') message_pattern = re.compile('\"(.+)\"') # Lambda execution starts here def handler(event, context): for record in event['Records']: # Get the bucket name and key for the new file bucket = record['s3']['bucket']['name'] key = record['s3']['object']['key'] # Get, read, and split the file into lines obj = s3.get_object(Bucket=bucket, Key=key) body = obj['Body'].read() lines = body.splitlines() # Match the regular expressions to each line and index the JSON for line in lines: line = line.decode("utf-8") ip = ip_pattern.search(line).group(1) timestamp = time_pattern.search(line).group(1) message = message_pattern.search(line).group(1) document = { "ip": ip, "timestamp": timestamp, "message": message } r = requests.post(url, auth=awsauth, json=document, headers=headers) ``` Modifiez les variables des champs `region` et `host`. 1. Si vous ne l'avez pas encore fait, [installez pip](https://pip.pypa.io/en/stable/installation/), puis installez les dépendances dans un nouveau répertoire `package` : ``` cd s3-to-opensearch pip install --target ./package requests pip install --target ./package requests_aws4auth ``` [Boto3](https://aws.amazon.com/sdk-for-python/) est installé dans tous les environnements d'exécution Lambda et vous n'avez pas besoin de l'inclure dans votre package de déploiement. 1. Empaquetez le code d'application et les dépendances : ``` cd package zip -r ../lambda.zip . cd .. zip -g lambda.zip sample.py ``` ## Créer la fonction Lambda Après avoir créé le package de déploiement, vous pouvez créer la fonction Lambda. Lorsque vous créez une fonction, choisissez un nom, une exécution (par exemple, Python 3.8) et un rôle IAM. Le rôle IAM définit les autorisations pour votre fonction. Pour obtenir des instructions détaillées, consultez [Création d'une fonction Lambda à l'aide de la console](https://docs.aws.amazon.com/lambda/latest/dg/get-started-create-function.html) dans le *Guide du développeur AWS Lambda *. Cet exemple suppose que vous utilisez la console. Choisissez Python 3.9 et un rôle doté des autorisations de lecture S3 et des autorisations d'écriture du OpenSearch service, comme illustré dans la capture d'écran suivante : ![\[Exemple de configuration d'une fonction Lambda\]](http://docs.aws.amazon.com/fr_fr/opensearch-service/latest/developerguide/images/lambda-function.png) Une fois que vous avez créé la fonction, vous devez ajouter un déclencheur. Pour cet exemple, nous voulons que le code s'exécute chaque fois qu'un fichier journal arrive dans un compartiment S3 : 1. Choisissez **Add trigger (Ajouter un déclencheur)** et sélectionnez **S3**. 1. Choisissez votre compartiment. 1. Pour **Event type (Type d'événement)**, choisissez **PUT**. 1. Pour **Préfixe**, tapez `logs/`. 1. Pour **Suffixe**, tapez `.log`. 1. Acceptez l'avertissement d'invocation récursive et choisissez **Add (Ajouter)**. Enfin, vous pouvez charger votre package de déploiement : 1. Choisissez **Upload from (Charger à partir de)** et **.zip file (Fichier .zip)**, puis suivez les invites pour charger votre package de déploiement. 1. Au terme du chargement, modifiez les **Paramètres d'exécution** et remplacez le **Gestionnaire** par `sample.handler`. Ce paramètre indique à Lambda le fichier (`sample.py`) et la méthode (`handler`) à exécuter après un déclencheur. À ce stade, vous disposez d'un ensemble complet de ressources : un compartiment pour les fichiers journaux, une fonction qui s'exécute chaque fois qu'un fichier journal est ajouté au compartiment, du code qui effectue l'analyse et l'indexation, et un domaine de OpenSearch service pour la recherche et la visualisation. ## Test de la fonction Lambda Après avoir créé la fonction, vous pouvez la tester en chargeant un fichier dans le compartiment Amazon S3. Créez un fichier nommé `sample.log` en utilisant les exemples de lignes de journal suivants : ``` 12.345.678.90 - [10/Oct/2000:13:55:36 -0700] "PUT /some-file.jpg" 12.345.678.91 - [10/Oct/2000:14:56:14 -0700] "GET /some-file.jpg" ``` Chargez le fichier dans le dossier `logs` de votre compartiment S3. Pour obtenir des instructions, consultez [Charger un objet dans votre compartiment](https://docs.aws.amazon.com/AmazonS3/latest/userguide/PuttingAnObjectInABucket.html) dans le *Guide de l'utilisateur Amazon Simple Storage Service*. Utilisez ensuite la console OpenSearch de service ou OpenSearch les tableaux de bord pour vérifier que l'`lambda-s3-index`index contient deux documents. Vous pouvez également effectuer une requête de recherche standard : ``` GET https://domain-name/lambda-s3-index/_search?pretty { "hits" : { "total" : 2, "max_score" : 1.0, "hits" : [ { "_index" : "lambda-s3-index", "_type" : "_doc", "_id" : "vTYXaWIBJWV_TTkEuSDg", "_score" : 1.0, "_source" : { "ip" : "12.345.678.91", "message" : "GET /some-file.jpg", "timestamp" : "10/Oct/2000:14:56:14 -0700" } }, { "_index" : "lambda-s3-index", "_type" : "_doc", "_id" : "vjYmaWIBJWV_TTkEuCAB", "_score" : 1.0, "_source" : { "ip" : "12.345.678.90", "message" : "PUT /some-file.jpg", "timestamp" : "10/Oct/2000:13:55:36 -0700" } } ] } } ``` # Chargement de données de streaming à partir d'Amazon Kinesis Data Streams Vous pouvez charger des données de streaming depuis Kinesis Data Streams OpenSearch vers Service. Les nouvelles données qui arrivent dans le flux de données déclenchent l'envoi d'une notification d'événement à Lambda, qui exécute alors votre code personnalisé pour effectuer l'indexation. Cette section inclut des exemples de code Python simple. ## Conditions préalables Avant de poursuivre, vous devez disposer des ressources suivantes. | Prérequis | Description | | --- | --- | | Flux de données Amazon Kinesis | Source d'événement de votre fonction Lambda. Pour en savoir plus, consultez [Kinesis Data Streams](https://docs.aws.amazon.com/kinesis/latest/dev/amazon-kinesis-streams.html). | | OpenSearch Domaine de service | Destination des données après leur traitement par votre fonction Lambda. Pour plus d'informations, consultez [Création de domaines OpenSearch de service](createupdatedomains.md#createdomains) | | Rôle IAM | Ce rôle doit disposer d'autorisations OpenSearch Service, Kinesis et Lambda de base, telles que les suivantes : JSON **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "es:ESHttpPost", "es:ESHttpPut", "logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents", "kinesis:GetShardIterator", "kinesis:GetRecords", "kinesis:DescribeStream", "kinesis:ListStreams" ], "Resource": "*" } ] } ``` Le rôle doit avoir la relation d'approbation suivante : JSON **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "Service": "lambda.amazonaws.com" }, "Action": "sts:AssumeRole" } ] } ``` Pour en savoir plus, consultez [Création de rôles IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_create.html) dans le *Guide de l'utilisateur IAM*. | # Créer la fonction Lambda Suivez les instructions fournies dans [Créer le package de déploiement Lambda](integrations-s3-lambda.md#integrations-s3-lambda-deployment-package), mais créez un répertoire nommé `kinesis-to-opensearch` et utilisez le code suivant pour `sample.py` : ``` import base64 import boto3 import json import requests from requests_aws4auth import AWS4Auth region = '' # e.g. us-west-1 service = 'es' credentials = boto3.Session().get_credentials() awsauth = AWS4Auth(credentials.access_key, credentials.secret_key, region, service, session_token=credentials.token) host = '' # the OpenSearch Service domain, e.g. https://search-mydomain.us-west-1.es.amazonaws.com index = 'lambda-kine-index' datatype = '_doc' url = host + '/' + index + '/' + datatype + '/' headers = { "Content-Type": "application/json" } def handler(event, context): count = 0 for record in event['Records']: id = record['eventID'] timestamp = record['kinesis']['approximateArrivalTimestamp'] # Kinesis data is base64-encoded, so decode here message = base64.b64decode(record['kinesis']['data']) # Create the JSON document document = { "id": id, "timestamp": timestamp, "message": message } # Index the document r = requests.put(url + id, auth=awsauth, json=document, headers=headers) count += 1 return 'Processed ' + str(count) + ' items.' ``` Modifiez les variables des champs `region` et `host`. Si vous ne l'avez pas encore fait, [installez pip](https://pip.pypa.io/en/stable/installation/), puis utilisez les commandes suivantes pour installer vos dépendances : ``` cd kinesis-to-opensearch pip install --target ./package requests pip install --target ./package requests_aws4auth ``` Ensuite, suivez les instructions fournies dans [Créer la fonction Lambda](integrations-s3-lambda.md#integrations-s3-lambda-create), mais spécifiez le rôle IAM issu de [Conditions préalables](integrations-kinesis.md#integrations-kinesis-lambda-prereq) et les paramètres suivants pour le déclencheur : + **Flux Kinesis** : votre flux Kinesis + **Taille de lot** : 100 + **Position de départ** : horizon Trim Pour en savoir plus, consultez [Présentation d'Amazon Kinesis Data Streams](https://docs.aws.amazon.com/streams/latest/dev/working-with-kinesis.html) dans le*Guide du développeur Amazon Kinesis Data Streams*. À ce stade, vous disposez d'un ensemble complet de ressources : un flux de données Kinesis, une fonction qui s'exécute une fois que le flux reçoit de nouvelles données et indexe ces données, et un domaine de OpenSearch service pour la recherche et la visualisation. # Tester la fonction Lambda Après avoir créé la fonction, vous pouvez la tester en ajoutant un nouvel enregistrement dans le flux de données à l'aide de l' AWS CLI : ``` aws kinesis put-record --stream-name test --data "My test data." --partition-key partitionKey1 --region us-west-1 ``` Utilisez ensuite la console OpenSearch de service ou OpenSearch les tableaux de bord pour vérifier qu'il `lambda-kine-index` contient un document. Vous pouvez également utiliser la demande suivante : ``` GET https://domain-name/lambda-kine-index/_search { "hits" : [ { "_index": "lambda-kine-index", "_type": "_doc", "_id": "shardId-000000000000:49583511615762699495012960821421456686529436680496087042", "_score": 1, "_source": { "timestamp": 1523648740.051, "message": "My test data.", "id": "shardId-000000000000:49583511615762699495012960821421456686529436680496087042" } } ] } ``` # Chargement de données de streaming à partir d'Amazon DynamoDB Vous pouvez l'utiliser AWS Lambda pour envoyer des données vers votre domaine de OpenSearch service depuis Amazon DynamoDB. Les nouvelles données qui arrivent dans la table de la base de données déclenchent l'envoi d'une notification d'événement à Lambda, qui exécute ensuite votre code personnalisé pour effectuer l'indexation. ## Conditions préalables Avant de poursuivre, vous devez disposer des ressources suivantes. | Prérequis | Description | | --- | --- | | Tableau DynamoDB | Le tableau contient vos données sources. Pour plus d'informations, consultez [Opérations de base sur les tables DynamoDB](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/WorkingWithTables.Basics.html)dans le *Guide du développeur Amazon DynamoDB*.La table doit résider dans la même région que votre domaine de OpenSearch service et avoir un flux défini sur **Nouvelle image**. Pour en savoir plus, consultez [Activation d'un flux](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html#Streams.Enabling). | | OpenSearch Domaine de service | Destination des données après leur traitement par votre fonction Lambda. Pour de plus amples informations, veuillez consulter [Création de domaines OpenSearch de service](createupdatedomains.md#createdomains). | | Rôle IAM | Ce rôle doit disposer des autorisations d'exécution de base de OpenSearch service, DynamoDB et Lambda, telles que les suivantes : JSON **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "es:ESHttpPost", "es:ESHttpPut", "dynamodb:DescribeStream", "dynamodb:GetRecords", "dynamodb:GetShardIterator", "dynamodb:ListStreams", "logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents" ], "Resource": "*" } ] } ``` Le rôle doit avoir la relation d'approbation suivante : JSON **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "Service": "lambda.amazonaws.com" }, "Action": "sts:AssumeRole" } ] } ``` Pour en savoir plus, consultez [Création de rôles IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_create.html) dans le *Guide de l'utilisateur IAM*. | ## Créer la fonction Lambda Suivez les instructions fournies dans [Créer le package de déploiement Lambda](integrations-s3-lambda.md#integrations-s3-lambda-deployment-package), mais créez un répertoire nommé `ddb-to-opensearch` et utilisez le code suivant pour `sample.py` : ``` import boto3 import requests from requests_aws4auth import AWS4Auth region = '' # e.g. us-east-1 service = 'es' credentials = boto3.Session().get_credentials() awsauth = AWS4Auth(credentials.access_key, credentials.secret_key, region, service, session_token=credentials.token) host = '' # the OpenSearch Service domain, e.g. https://search-mydomain.us-west-1.es.amazonaws.com index = 'lambda-index' datatype = '_doc' url = host + '/' + index + '/' + datatype + '/' headers = { "Content-Type": "application/json" } def handler(event, context): count = 0 for record in event['Records']: # Get the primary key for use as the OpenSearch ID id = record['dynamodb']['Keys']['id']['S'] if record['eventName'] == 'REMOVE': r = requests.delete(url + id, auth=awsauth) else: document = record['dynamodb']['NewImage'] r = requests.put(url + id, auth=awsauth, json=document, headers=headers) count += 1 return str(count) + ' records processed.' ``` Modifiez les variables des champs `region` et `host`. Si vous ne l'avez pas encore fait, [installez pip](https://pip.pypa.io/en/stable/installation/), puis utilisez les commandes suivantes pour installer vos dépendances : ``` cd ddb-to-opensearch pip install --target ./package requests pip install --target ./package requests_aws4auth ``` Ensuite, suivez les instructions fournies dans [Créer la fonction Lambda](integrations-s3-lambda.md#integrations-s3-lambda-create), mais spécifiez le rôle IAM issu de [Conditions préalables](#integrations-dynamodb-prereq) et les paramètres suivants pour le déclencheur : + **Table** : votre table DynamoDB + **Taille de lot** : 100 + **Position de départ** : horizon Trim Pour en savoir plus, consultez [Traitement des nouveaux éléments avec DynamoDB Streams et Lambda](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.Lambda.Tutorial.html) dans le *Guide du développeur Amazon DynamoDB*. À ce stade, vous disposez d'un ensemble complet de ressources : une table DynamoDB pour vos données source, un flux DynamoDB contenant les modifications apportées à la table, une fonction qui s'exécute après les modifications de vos données source et indexe ces modifications, et un domaine de service pour la recherche et la visualisation. OpenSearch ## Test de la fonction Lambda Après avoir créé la fonction, vous pouvez la tester en ajoutant un nouvel élément dans la table DynamoDB à l'aide de l'interface AWS CLI : ``` aws dynamodb put-item --table-name test --item '{"director": {"S": "Kevin Costner"},"id": {"S": "00001"},"title": {"S": "The Postman"}}' --region us-west-1 ``` Utilisez ensuite la console OpenSearch de service ou OpenSearch les tableaux de bord pour vérifier qu'il `lambda-index` contient un document. Vous pouvez également utiliser la demande suivante : ``` GET https://domain-name/lambda-index/_doc/00001 { "_index": "lambda-index", "_type": "_doc", "_id": "00001", "_version": 1, "found": true, "_source": { "director": { "S": "Kevin Costner" }, "id": { "S": "00001" }, "title": { "S": "The Postman" } } } ``` # Chargement de données de streaming depuis Amazon Data Firehose Firehose prend en charge le OpenSearch service en tant que destination de livraison. Pour savoir comment charger des données de streaming dans OpenSearch Service, consultez les sections [Creating a Kinesis Data Firehose](https://docs.aws.amazon.com/firehose/latest/dev/basic-create.html) Delivery [Stream OpenSearch et Choose Service for](https://docs.aws.amazon.com/firehose/latest/dev/create-destination.html#create-destination-elasticsearch) Your Destination dans *le manuel Amazon Data Firehose* Developer Guide. Avant de charger des données dans OpenSearch Service, vous devrez peut-être effectuer des transformations sur les données. Pour en savoir plus sur l'utilisation des fonctions Lambda pour effectuer cette tâche, consultez [Transformation de données Amazon Kinesis Data Firehose](https://docs.aws.amazon.com/firehose/latest/dev/data-transformation.html) dans le même guide. Lorsque vous configurez un flux de diffusion, Firehose propose un rôle IAM « en un clic » qui lui donne l'accès aux ressources dont il a besoin pour envoyer des données au OpenSearch Service, sauvegarder des données sur Amazon S3 et transformer des données à l'aide de Lambda. En raison de la complexité du processus de création manuelle d'un tel rôle, nous vous recommandons d'utiliser le rôle fourni. # Chargement de données de streaming depuis Amazon CloudWatch Vous pouvez charger des données de streaming depuis CloudWatch Logs vers votre domaine OpenSearch de service à l'aide d'un abonnement CloudWatch Logs. Pour plus d'informations sur les CloudWatch abonnements Amazon, consultez la section [Traitement en temps réel des données de journal dans le cadre des abonnements](https://docs.aws.amazon.com/AmazonCloudWatch/latest/DeveloperGuide/Subscriptions.html). Pour obtenir des informations de configuration, consultez la section [Streaming CloudWatch Logs to Amazon OpenSearch Service](https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/CWL_OpenSearch_Stream.html) dans le manuel *Amazon CloudWatch Developer Guide*. ## Chargement de données de streaming depuis AWS IoT Vous pouvez envoyer des données à l' AWS IoT aide de [règles](https://docs.aws.amazon.com/iot/latest/developerguide/iot-rules.html). Pour en savoir plus, consultez l'[OpenSearch](https://docs.aws.amazon.com/iot/latest/developerguide/opensearch-rule-action.html)action décrite dans le *guide du AWS IoT développeur*.