Cible du Catalogue Configuration de votre compte pour les notifications d'événements Amazon S3

Accélération des analyseurs à l'aide des notifications d'événements Amazon S3

Au lieu de répertorier les objets d'une cible Amazon S3 ou de Catalogue de données, vous pouvez configurer le Crawler pour qu'il utilise les événements Amazon S3 pour trouver les modifications éventuelles. Cette fonction améliore le temps de nouvelle analyse en utilisant les événements Amazon S3 pour identifier les changements entre deux analyses. Ce processus s'effectue en répertoriant tous les fichiers du sous-dossier qui a déclenché l'événement au lieu de répertorier la cible Amazon S3 ou du Catalogue de données complète.

La première analyse répertorie tous les objets Amazon S3 de la cible. Après la première analyse réussie, vous pouvez choisir de faire une nouvelle analyse manuellement ou selon un calendrier défini. L'crawler indiquera uniquement les objets de ces événements au lieu de répertorier tous les objets.

Passer à un crawler basé sur les événements Amazon S3 a les avantages suivants :

Une nouvelle analyse plus rapide, car la liste de tous les objets de la cible n'est pas nécessaire ; toutefois, la liste de dossiers spécifiques est effectuée là où des objets sont ajoutés ou supprimés.
Une réduction du coût global d'analyse étant donné que la liste de dossiers spécifiques est effectuée lorsque des objets sont ajoutés ou supprimés.

L'analyse des événements Amazon S3 s'exécute en consommant des événements Amazon S3 depuis la file d'attente SQS en fonction de la planification de l'crawler. Il n'y aura aucun coût s'il n'y a pas d'événements dans la file d'attente. Les événements Amazon S3 peuvent être configurés pour accéder directement à la file d'attente SQS ou, dans les cas où plusieurs consommateurs ont besoin du même événement, une combinaison de SNS et de SQS. Pour plus d’informations, consultez Configuration de votre compte pour les notifications d'événements Amazon S3.

Après avoir créé et configuré le Crawler en mode événement, la première analyse s'exécute en mode de liste en répertoriant la liste de la cible Amazon S3 ou du Catalogue de données complète. Le journal suivant confirme l'opération de l'analyse en consommant des événements Amazon S3 après la première analyse réussie : « L'analyse est exécutée en consommant des événements Amazon S3. »

Après avoir créé l'analyse des événements Amazon S3 et mis à jour les propriétés de l'crawler susceptibles d'avoir un impact sur l'analyse, l'analyse fonctionne en mode liste et le journal suivant est ajouté : « L'analyse n'est pas exécutée en mode événement S3 ».

Note

Le nombre maximum de messages à consommer est de 10 000 messages par crawl.

Cible du Catalogue

Lorsque la cible est le Catalogue de données, le Crawler met à jour les tables existantes du Catalogue de données avec les modifications (par exemple, des partitions supplémentaires dans une table).

Rubriques

Configuration de votre compte pour les notifications d'événements Amazon S3

Cette section décrit comment configurer votre compte pour les notifications d'événements Amazon S3 et fournit des instructions pour ce faire à l'aide d'un script ou de la AWS Glue console.

Prérequis

Réalisez les tâches de configuration suivantes. Notez que les valeurs entre parenthèses font référence aux paramètres configurables du script.

Créez un compartiment Amazon S3 (s3_bucket_name).
Identifiez une cible de l'crawler (folder_name, comme « test1 ») qui est un chemin dans le compartiment identifié.
Préparez un nom d'crawler (crawler_name)
Préparez un nom de rubrique SNS (sns_topic_name) qui pourrait être le même que le nom de l'crawler.
Préparez la AWS région dans laquelle le robot doit s'exécuter et où le compartiment S3 existe (region).
Préparez éventuellement une adresse e-mail si l'e-mail est utilisé pour obtenir les événements Amazon S3 (subscribing_email).

Vous pouvez également utiliser la CloudFormation pile pour créer vos ressources. Procédez comme suit :

Lancez votre CloudFormation stack dans l'est des États-Unis (Virginie du Nord) :
Sous Paramètres, saisissez un nom pour votre compartiment Amazon S3 (incluez votre numéro de compte).
Sélectionnez I acknowledge that AWS CloudFormation might create IAM resources with custom names.
Sélectionnez Create stack.

Limites:

Seule une cible unique est prise en charge par le Crawler, que ce soit pour les cibles Amazon S3 ou du Catalogue de données.
Un SQS sur un VPC privé n'est pas pris en charge.
L'échantillonnage Amazon S3 n'est pas pris en charge.
La cible du Crawler doit être un dossier pour une cible Amazon S3, ou une ou plusieurs tables de Catalogue de données AWS Glue pour une cible de Catalogue de données.
Le caractère générique du chemin « tout » n'est pas pris en charge : s3://%
Pour une cible de Catalogue de données, toutes les tables du catalogue doivent pointer vers le même compartiment Amazon S3 pour le mode événement Amazon S3.
Pour une cible de Catalogue de données, une table de catalogue ne doit pas pointer vers un emplacement Amazon S3 au format Delta Lake (contenant des dossiers _symlink ou vérifiant le InputFormat de la table du catalogue).

Pour utiliser l'crawler basé sur les événements Amazon S3, vous devez activer la notification d'événements sur le compartiment S3 avec des événements filtrés à partir du préfixe identique à la cible et au magasin S3 dans SQS. Vous pouvez configurer SQS et la notification d'événements via la console en suivant les étapes de Démonstration : configuration d'un compartiment en vue des notifications ou en utilisant le Script pour la génération de SQS et la configuration des événements Amazon S3 à partir de la cible.

Politique SQS

Ajoutez la politique SQS suivante qui doit être attachée au rôle utilisé par l'crawler.


{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "VisualEditor0",
            "Effect": "Allow",
            "Action": [
                "sqs:DeleteMessage",
                "sqs:GetQueueUrl",
                "sqs:ListDeadLetterSourceQueues",                
                "sqs:ReceiveMessage",
                "sqs:GetQueueAttributes",
                "sqs:ListQueueTags",
                "sqs:SetQueueAttributes",
                "sqs:PurgeQueue"
            ],
            "Resource": "arn:aws:sqs:{region}:{accountID}:cfn-sqs-queue"
        }
    ]
}

Script pour la génération de SQS et la configuration des événements Amazon S3 à partir de la cible

Une fois que les conditions préalables sont satisfaites, vous pouvez exécuter le script Python suivant pour créer le SQS. Remplacez les paramètres configurables par les noms préparés à partir des conditions préalables.

Note

Après avoir exécuté le script, connectez-vous à la console SQS pour trouver l'ARN du SQS créé.

Amazon SQS définit un délai de visibilité, une période au cours de laquelle Amazon SQS empêche les autres consommateurs de recevoir et de traiter le message. Définissez le délai de visibilité approximativement égal à l'exécution de l'analyse.


#!venv/bin/python
import boto3
import botocore
 
#---------Start : READ ME FIRST ----------------------#
# 1. Purpose of this script is to create the SQS, SNS and enable S3 bucket notification.
#     The following are the operations performed by the scripts:
#      a. Enable S3 bucket notification to trigger 's3:ObjectCreated:' and 's3:ObjectRemoved:' events.
#      b. Create SNS topic for fan out.
#      c. Create SQS queue for saving events which will be consumed by the crawler.
#          SQS Event Queue ARN will be used to create the crawler after running the script.
# 2. This script does not create the crawler.
# 3. SNS topic is created to support FAN out of S3 events. If S3 event is also used by another
#    purpose, SNS topic created by the script can be used.
# 1. Creation of bucket is an optional step.
#    To create a bucket set create_bucket variable to true.
# 2. The purpose of crawler_name is to easily locate the SQS/SNS.
#     crawler_name is used to create SQS and SNS with the same name as crawler.
# 3. 'folder_name' is the target of crawl inside the specified bucket 's3_bucket_name'
#
#---------End : READ ME FIRST ------------------------#
 
 
#--------------------------------#
# Start : Configurable settings  #
#--------------------------------#
 
#Create
region = 'us-west-2'
s3_bucket_name = 's3eventtestuswest2'
folder_name = "test"
crawler_name = "test33S3Event"
sns_topic_name = crawler_name
sqs_queue_name = sns_topic_name
create_bucket = False
 
#-------------------------------#
# End : Configurable settings   #
#-------------------------------#
 
# Define aws clients
dev = boto3.session.Session(profile_name='myprofile')
boto3.setup_default_session(profile_name='myprofile')
s3 = boto3.resource('s3', region_name=region)
sns = boto3.client('sns', region_name=region)
sqs = boto3.client('sqs', region_name=region)
client = boto3.client("sts")
account_id = client.get_caller_identity()["Account"]
queue_arn = ""
 
 
def print_error(e):
    print(e.message + ' RequestId: ' + e.response['ResponseMetadata']['RequestId'])
 
def create_s3_bucket(bucket_name, client):
    bucket = client.Bucket(bucket_name)
    try:
        if not create_bucket:
            return True
        response = bucket.create(
            ACL='private',
            CreateBucketConfiguration={
                'LocationConstraint': region
            },
        )
        return True
    except botocore.exceptions.ClientError as e:
        print_error(e)
        if 'BucketAlreadyOwnedByYou' in e.message:  # we own this bucket so continue
            print('We own the bucket already. Lets continue...')
            return True
    return False
 
def create_s3_bucket_folder(bucket_name, client, directory_name):
    s3.put_object(Bucket=bucket_name, Key=(directory_name + '/'))
 
def set_s3_notification_sns(bucket_name, client, topic_arn):
    bucket_notification = client.BucketNotification(bucket_name)
    try:
 
        response = bucket_notification.put(
            NotificationConfiguration={
                'TopicConfigurations': [
                    {
                        'Id' : crawler_name,
                        'TopicArn': topic_arn,
                        'Events': [
                            's3:ObjectCreated:*',
                            's3:ObjectRemoved:*',
 
                        ],
                        'Filter' :  {'Key': {'FilterRules': [{'Name': 'prefix', 'Value': folder_name}]}}
                    },
                ]
            }
        )
        return True
    except botocore.exceptions.ClientError as e:
        print_error(e)
    return False
 
 
def create_sns_topic(topic_name, client):
    try:
        response = client.create_topic(
            Name=topic_name
        )
        return response['TopicArn']
    except botocore.exceptions.ClientError as e:
        print_error(e)
    return None
 
def set_sns_topic_policy(topic_arn, client, bucket_name):
    try:
        response = client.set_topic_attributes(
            TopicArn=topic_arn,
            AttributeName='Policy',
            AttributeValue='''{
              "Version": "2008-10-17",
              "Id": "s3-publish-to-sns",
              "Statement": [{
                  "Effect": "Allow",
                  "Principal": { "AWS" : "*" },
                  "Action": [ "SNS:Publish" ],
                  "Resource": "%s",
                  "Condition": {
                    "StringEquals": {
                        "AWS:SourceAccount": "%s"
                      },
                      "ArnLike": {
                          "aws:SourceArn": "arn:aws:s3:*:*:%s"
                      }
                  }
              }]
  }''' % (topic_arn, account_id, bucket_name)
        )
        return True
    except botocore.exceptions.ClientError as e:
        print_error(e)
 
    return False
 
 
def subscribe_to_sns_topic(topic_arn, client, protocol, endpoint):
    try:
        response = client.subscribe(
            TopicArn=topic_arn,
            Protocol=protocol,
            Endpoint=endpoint
        )
        return response['SubscriptionArn']
    except botocore.exceptions.ClientError as e:
        print_error(e)
    return None
 
 
def create_sqs_queue(queue_name, client):
    try:
        response = client.create_queue(
            QueueName=queue_name,
        )
        return response['QueueUrl']
    except botocore.exceptions.ClientError as e:
        print_error(e)
    return None
 
 
def get_sqs_queue_arn(queue_url, client):
    try:
        response = client.get_queue_attributes(
            QueueUrl=queue_url,
            AttributeNames=[
                'QueueArn',
            ]
        )
        return response['Attributes']['QueueArn']
    except botocore.exceptions.ClientError as e:
        print_error(e)
    return None
 
def set_sqs_policy(queue_url, queue_arn, client, topic_arn):
    try:
        response = client.set_queue_attributes(
            QueueUrl=queue_url,
            Attributes={
                'Policy': '''{
                  "Version": "2012-10-17",
                  "Id": "AllowSNSPublish",
                  "Statement": [
                    {
                      "Sid": "AllowSNSPublish01",
                      "Effect": "Allow",
                      "Principal": "*",
                      "Action": "SQS:SendMessage",
                      "Resource": "%s",
                      "Condition": {
                        "ArnEquals": {
                          "aws:SourceArn": "%s"
                        }
                      }
                    }
                  ]
}''' % (queue_arn, topic_arn)
            }
        )
        return True
    except botocore.exceptions.ClientError as e:
        print_error(e)
    return False
 
 
if __name__ == "__main__":
    print('Creating S3 bucket %s.' % s3_bucket_name)
    if create_s3_bucket(s3_bucket_name, s3):
        print('\nCreating SNS topic %s.' % sns_topic_name)
        topic_arn = create_sns_topic(sns_topic_name, sns)
        if topic_arn:
            print('SNS topic created successfully: %s' % topic_arn)
 
            print('Creating SQS queue %s' % sqs_queue_name)
            queue_url = create_sqs_queue(sqs_queue_name, sqs)
            if queue_url is not None:
                print('Subscribing sqs queue with sns.')
                queue_arn = get_sqs_queue_arn(queue_url, sqs)
                if queue_arn is not None:
                    if set_sqs_policy(queue_url, queue_arn, sqs, topic_arn):
                        print('Successfully configured queue policy.')
                        subscription_arn = subscribe_to_sns_topic(topic_arn, sns, 'sqs', queue_arn)
                        if subscription_arn is not None:
                            if 'pending confirmation' in subscription_arn:
                                print('Please confirm SNS subscription by visiting the subscribe URL.')
                            else:
                                print('Successfully subscribed SQS queue: ' + queue_arn)
                        else:
                            print('Failed to subscribe SNS')
                    else:
                        print('Failed to set queue policy.')
                else:
                    print("Failed to get queue arn for %s" % queue_url)
            # ------------ End subscriptions to SNS topic -----------------
 
            print('\nSetting topic policy to allow s3 bucket %s to publish.' % s3_bucket_name)
            if set_sns_topic_policy(topic_arn, sns, s3_bucket_name):
                print('SNS topic policy added successfully.')
                if set_s3_notification_sns(s3_bucket_name, s3, topic_arn):
                    print('Successfully configured event for S3 bucket %s' % s3_bucket_name)
                    print('Create S3 Event Crawler using SQS ARN %s' % queue_arn)
                else:
                    print('Failed to configure S3 bucket notification.')
            else:
                print('Failed to add SNS topic policy.')
        else:
            print('Failed to create SNS topic.')

Configuration d'un Crawler pour les notifications d'événements Amazon S3 à l'aide de la console (cible Amazon S3)

Pour configurer un Crawler pour les notifications d'événements Amazon S3 à l'aide de la console AWS Glue pour une cible Amazon S3 :

Définissez les propriétés de votre crawler. Pour plus d'informations, consultez la rubrique Configuration des options de configuration du crawler dans la console AWS Glue.
Dans la section Configuration de source de données, la question Is your data already mapped to AWS Glue tables? (Vos données sont-elles déjà mappées aux tables ?) vous est posée.

Par défaut, la réponse Not yet (Pas encore) est déjà sélectionnée. Laissez cette réponse par défaut, car vous utilisez une source de données Amazon S3 et les données ne sont pas encore mappées aux tables AWS Glue.
Dans la section Data sources (Sources de données), choisissez Add a data source (Ajouter une source de données).
Dans le modal Add data source (Ajouter une source de données), configurez la source de données Amazon S3 :
- Data source (Source de données) : Amazon S3 est sélectionné par défaut.
- Network connection (Connexion réseau) (Facultatif) : sélectionnez Add new connection (Ajouter une nouvelle connexion).
- Location of Amazon S3 data (Emplacement des données Amazon S3) : par défaut, l'option In this account (Dans ce compte) est sélectionnée.
- Amazon S3 path (Chemin d'accès Amazon S3) : spécifiez le chemin d'accès Amazon S3 où les dossiers et les fichiers sont analysés.
- Subsequent crawler runs (Exécutions ultérieures du crawler) : choisissez Crawl based on events (Analyse en fonction des événements) pour utiliser les notifications d'événements Amazon S3 pour votre crawler.
- Include SQS ARN (Inclure un ARN SQS) : spécifiez les paramètres du magasin de données, y compris l'ARN SQS valide. (Par exemple, arn:aws:sqs:region:account:sqs.)
- Include dead-letter SQS ARN (Inclure un ARN SQS de lettres mortes) (Facultatif) : spécifiez un ARN SQS de lettres mortes Amazon valide. (Par exemple, arn:aws:sqs:region:account:deadLetterQueue.)
- Choisissez Add an Amazon S3 data source (Ajouter une source de données Amazon S3).

Configuration d'un robot d'exploration pour les notifications d'événements Amazon S3 à l'aide du AWS CLI

Voici un exemple d' AWS CLI appel Amazon S3 pour créer des files d'attente SQS et configurer des notifications d'événements sur le compartiment cible Amazon S3.



S3 Event AWS CLI
aws sqs create-queue --queue-name MyQueue --attributes file://create-queue.json
create-queue.json
```
{
    "Policy": {
        "Version": "2012-10-17",
        "Id": "example-ID",
        "Statement": [
            {
                "Sid": "example-statement-ID",
                "Effect": "Allow",
                "Principal": {
                    "Service": "s3.amazonaws.com"
                },
                "Action": [
                    "SQS:SendMessage"
                ],
                "Resource": "SQS-queue-ARN",
                "Condition": {
                    "ArnLike": {
                        "aws:SourceArn": "arn:aws:s3:*:*:awsexamplebucket1"
                    },
                    "StringEquals": {
                        "aws:SourceAccount": "bucket-owner-account-id"
                    }
                }
            }
        ]
    }
}
```
aws s3api put-bucket-notification-configuration --bucket customer-data-pdx --notification-configuration file://s3-event-config.json
s3-event-config.json
```
{
    "QueueConfigurations": [
        {
          "Id": "s3event-sqs-queue",
          "QueueArn": "arn:aws:sqs:{region}:{account}:queuename",
          "Events": [
                "s3:ObjectCreated:*",
                "s3:ObjectRemoved:*"
          ],
          "Filter": {
              "Key": {
                  "FilterRules": [
                      {
                          "Name": "Prefix",
                          "Value": "/json"
                      }
                  ]
              }
          }
        }
    ]
}
```
Create Crawler:

Configuration d'un Crawler pour les notifications d'événements Amazon S3 à l'aide de la console (cible du Catalogue de données)

Lorsque vous avez une cible de catalogue, configurez un Crawler pour les notifications d'événements Amazon S3 à l'aide de la console AWS Glue :

Définissez les propriétés de votre crawler. Pour plus d'informations, consultez la rubrique Configuration des options de configuration du crawler dans la console AWS Glue.
Dans la section Configuration de source de données, la question Is your data already mapped to AWS Glue tables? (Vos données sont-elles déjà mappées aux tables ?) vous est posée.

Sélectionnez Yes (Oui) pour choisir les tables existantes de votre Catalogue de données comme source de données.
Dans la section Glue tables (Tables Glue), choisissez Add tables (Ajouter des tables).
Dans le modal Add table (Ajouter une table), configurez la base de données et les tables :
- Network connection (Connexion réseau) (Facultatif) : sélectionnez Add new connection (Ajouter une nouvelle connexion).
- Database (Base de données) : sélectionnez une base de données dans le Catalogue de données.
- Tables : sélectionnez une ou plusieurs tables de cette base de données dans le Catalogue de données.
- Subsequent crawler runs (Exécutions ultérieures du crawler) : choisissez Crawl based on events (Analyse en fonction des événements) pour utiliser les notifications d'événements Amazon S3 pour votre crawler.
- Include SQS ARN (Inclure un ARN SQS) : spécifiez les paramètres du magasin de données, y compris l'ARN SQS valide. (Par exemple, arn:aws:sqs:region:account:sqs.)
- Include dead-letter SQS ARN (Inclure un ARN SQS de lettres mortes) (Facultatif) : spécifiez un ARN SQS de lettres mortes Amazon valide. (Par exemple, arn:aws:sqs:region:account:deadLetterQueue.)
- Choisissez Confirmer.

Avertissement JavaScript est désactivé ou n'est pas disponible dans votre navigateur.

Pour que vous puissiez utiliser la documentation AWS, Javascript doit être activé. Vous trouverez des instructions sur les pages d'aide de votre navigateur.

Conventions de rédaction

Définition de l'option de configuration du Crawler de l'index de partition

Utilisation du chiffrement avec le crawler d'événements Amazon S3