Beschleunigung von Crawls mithilfe von Amazon S3-Ereignisbenachrichtigungen - AWS Glue
Catalog-ZielEinrichten eines Kontos für Amazon S3-Ereignisbenachrichtigungen

Die vorliegende Übersetzung wurde maschinell erstellt. Im Falle eines Konflikts oder eines Widerspruchs zwischen dieser übersetzten Fassung und der englischen Fassung (einschließlich infolge von Verzögerungen bei der Übersetzung) ist die englische Fassung maßgeblich.

Beschleunigung von Crawls mithilfe von Amazon S3-Ereignisbenachrichtigungen

Anstatt die Objekte aus einem Amazon S3- oder Data Catalog-Ziel aufzulisten, können Sie den Crawler so konfigurieren, dass er Amazon S3-Ereignisse verwendet, um Änderungen zu finden. Dieses Feature verbessert die Recrawl-Zeit, indem Amazon S3-Ereignisse verwendet werden, um die Änderungen zwischen zwei Crawls zu identifizieren, indem alle Dateien aus dem Unterordner aufgeführt werden, der das Ereignis ausgelöst hat, anstatt das vollständige Amazon S3- oder Data Calatog-Ziel aufzulisten.

Der erste Crawl listet alle Amazon S3-Objekte aus dem Ziel auf. Nach dem ersten erfolgreichen Crawl können Sie wählen, ob Sie manuell oder nach einem festgelegten Zeitplan recrawlen möchten. Der Crawler listet nur die Objekte dieser Ereignisse auf, anstatt alle Objekte aufzulisten.

Die Vorteile eines Umstiegs zu einem ereignisbasierten Crawler auf Amazon S3 sind:

  • Ein schnelleres Recrawl, da die Auflistung aller Objekte aus dem Ziel nicht erforderlich ist, stattdessen erfolgt die Auflistung bestimmter Ordner, in denen Objekte hinzugefügt oder gelöscht werden.

  • Eine Senkung der Gesamtdurchforstungskosten, da die Auflistung bestimmter Ordner erfolgt, in denen Objekte hinzugefügt oder gelöscht werden.

Die Amazon S3-Ereignisdurchforstung wird ausgeführt, indem Amazon S3-Ereignisse aus der SQS-Warteschlange basierend auf dem Crawler-Zeitplan verwendet werden. Es fallen keine Kosten an, wenn keine Ereignisse in der Warteschlange vorhanden sind. Amazon S3-Ereignisse können so konfiguriert werden, dass sie direkt in die SQS-Warteschlange bzw. in Fällen, in denen mehrere Verbraucher dasselbe Ereignis benötigen, in eine Kombination aus SNS und SQS gelangen. Weitere Informationen finden Sie unter Einrichten eines Kontos für Amazon S3-Ereignisbenachrichtigungen.

Nach dem Erstellen und Konfigurieren des Crawlers im Ereignismodus wird das erste Crawl im Listing-Modus ausgeführt, indem eine vollständige Auflistung des Amazon S3- oder Data Catalog-Ziels ausgeführt wird. Das folgende Protokoll bestätigt den Crawl-Vorgang durch den Verbrauch von Amazon S3-Ereignissen nach dem ersten erfolgreichen Crawl: „Der Crawl wird durch den Verbrauch von Amazon S3-Ereignissen ausgeführt.“

Nach dem Erstellen des Amazon S3-Ereigniscrawls und dem Aktualisieren der Crawler-Eigenschaften, die sich auf das Crawl auswirken können, wird das Crawl im List-Modus ausgeführt und das folgende Protokoll wird hinzugefügt: „Crawl läuft nicht im S3-Ereignismodus“.

Catalog-Ziel

Wenn das Ziel der Data Catalog ist, aktualisiert der Crawler die vorhandenen Tabellen im Data Catalog mit Änderungen (z. B. zusätzliche Partitionen in einer Tabelle).

Einrichten eines Kontos für Amazon S3-Ereignisbenachrichtigungen

In diesem Abschnitt wird beschrieben, wie Sie Ihr Konto für Amazon S3 S3-Ereignisbenachrichtigungen einrichten. Außerdem finden Sie Anweisungen dazu mithilfe eines Skripts oder der AWS Glue Konsole.

Voraussetzungen

Führen Sie die folgenden Einrichtungsaufgaben aus. Beachten Sie, dass die Werte in Klammern auf die konfigurierbaren Einstellungen aus dem Skript verweisen.

  1. Erstellen eines Amazon-S3-Buckets (s3_bucket_name).

  2. Identifizieren Sie ein Crawler-Ziel (folder_name, wie „test1"), was ein Pfad im identifizierten Bucket ist.

  3. Bereiten Sie einen Crawlernamen vor (crawler_name)

  4. Bereiten Sie einen SNS-Themennamen vor (sns_topic_name). Er könnte mit dem Crawlernamen identisch sein.

  5. Bereiten Sie die AWS Region vor, in der der Crawler ausgeführt werden soll und in der der S3-Bucket existiert (region).

  6. Bereiten Sie optional eine E-Mail-Adresse vor, wenn die Amazon S3-Ereignisse über E-Mail abgerufen werden (subscribing_email).

Sie können den CloudFormation Stack auch verwenden, um Ihre Ressourcen zu erstellen. Führen Sie folgende Schritte aus:

  1. Starten Sie Ihren CloudFormation Stack im Osten der USA (Nord-Virginia):

  2. Geben Sie unter Parameter einen Namen für Ihren Amazon-S3-Bucket ein (einschließlich Ihrer Kontonummer).

  3. Wählen Sie I acknowledge that AWS CloudFormation might create IAM resources with custom names.

  4. Wählen Sie Create stack.

Einschränkungen:

  • Nur ein einziges Ziel wird vom Crawler unterstützt, unabhängig davon, ob es sich um Amazon S3- oder Data Catalog-Ziele handelt.

  • SQS auf privater VPC wird nicht unterstützt.

  • Amazon S3-Probenahme wird nicht unterstützt.

  • Das Crawlerziel sollte ein Ordner für ein Amazon S3-Ziel oder eine oder mehrereAWS Glue Data Catalog-Tabellen für ein Data Catalog-Ziel sein.

  • Der Pfad-Platzhalter „alles“ wird nicht unterstützt: s3://%

  • Bei einem Data Catalog-Ziel sollten alle Katalogtabellen auf denselben Amazon S3-Bucket für den Amazon S3-Ereignismodus verweisen.

  • Bei einem Data Catalog-Ziel sollte eine Katalogtabelle nicht auf einen Amazon S3-Speicherort im Delta Lake-Format verweisen (der Ordner _symlink enthält oder das InputFormat der Katalogtabelle überprüft).

Um den ereignisbasierten Amazon S3-Crawler zu verwenden, sollten Sie die Ereignisbenachrichtigung auf dem S3-Bucket aktivieren, wobei Ereignisse aus dem Präfix gefiltert werden, was dem S3-Ziel und dem Speicher in SQS entspricht. Sie können SQS und Ereignisbenachrichtigungen über die Konsole einrichten, indem Sie die Schritte unter Exemplarische Vorgehensweise: Konfigurieren eines Buckets für Benachrichtigungen befolgen oder das Skript zum Generieren von SQS und Konfigurieren von Amazon S3-Ereignissen aus dem Ziel verwenden.

SQS-Richtlinie

Fügen Sie die folgende SQS-Richtlinie hinzu, die an die vom Crawler verwendete Rolle angehängt werden muss. 

{
    "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"
        }
    ]
}

Skript zum Generieren von SQS und Konfigurieren von Amazon S3-Ereignissen aus dem Ziel

Nachdem Sie sichergestellt haben, dass die Voraussetzungen erfüllt sind, können Sie das folgende Python-Skript ausführen, um das SQS zu erstellen. Ersetzen Sie die „Konfigurierbaren Einstellungen“ durch die Namen, die aus den Voraussetzungen erstellt wurden.

Anmerkung

Melden Sie sich nach dem Ausführen des Skripts bei der SQS-Konsole an, um den ARN des erstellten SQS zu finden.

Amazon SQS richtet eine Zeitbeschränkung für die Sichtbarkeit ein. Das ist ein Zeitraum, während dem Amazon SQS verhindert, dass andere Konsumenten eine Nachricht empfangen und verarbeiten. Stellen Sie die Zeitbeschränkung für die Sichtbarkeit ungefähr so wie die Crawl-Laufzeit ein.

#!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.')

Einrichten eines Crawlers für Amazon S3-Ereignisbenachrichtigungen mit der Konsole (Amazon S3-Ziel)

So richten Sie einen Crawler für Amazon S3-Ereignisbenachrichtigungen mit der AWS Glue-Konsole für ein Amazon S3-Ziel ein:

  1. Legen Sie Ihre Crawler-Eigenschaften fest. Weitere Informationen finden Sie unter Festlegen von Crawler-Konfigurationsoptionen auf der AWS Glue-Konsole.

  2. Im Abschnitt Data source configuration (Datenquellenkonfiguration) werden Sie Folgendes gefragt: Sind Ihre Daten bereits AWS Glue-Tabellen zugeordnet?

    Standardmäßig ist Not yet (Noch nicht) ausgewählt. Behalten Sie bei diesem die Standardeinstellungen bei, da Sie eine Amazon-S3-Datenquelle verwenden und die Daten noch keinen AWS Glue-Tabellen zugeordnet sind.

  3. Wählen Sie im Abschnitt Data sources (Datenquellen) Add a data source (Datenquelle hinzufügen) aus.

  4. Konfigurieren Sie im Modal Add a data source (Datenquelle hinzufügen) die Amazon-S3-Datenquelle:

    • Data source (Datenquelle): Standardmäßig ist Amazon S3 ausgewählt.

    • Network connection (Netzwerkverbindung) (Optional): Wählen Sie Add new connection (Neue Verbindung hinzufügen).

    • Location of Amazon S3 data (Speicherort der Amazon-S3-Daten): Standardmäßig ist In this account (In diesem Konto) ausgewählt.

    • Amazon S3 path (Amazon-S3-Pfad): Geben Sie den Amazon-S3-Pfad an, wo Ordner und Dateien gecrawlt werden.

    • Subsequent crawler runs (Nachfolgende Crawler-Ausführungen): Wählen Sie Crawl based on events (Crawling basierend auf Ereignissen) aus, um Amazon-S3-Ereignisbenachrichtigungen für Ihren Crawler zu verwenden.

    • SQS ARN hinzufügen: Geben Sie die Datenspeicherparameter einschließlich eines gültigen SQS ARN an. (Beispiel: arn:aws:sqs:region:account:sqs).

    • Dead-Letter SQS ARN hinzufügen (Optional): Geben Sie einen gültigen Amazon Dead-Letter SQS ARN an. (Beispiel: arn:aws:sqs:region:account:deadLetterQueue).

    • Wählen Sie Add an Amazon S3 data source (Amazon-S3-Datenquelle hinzufügen) aus.

Einrichtung eines Crawlers für Amazon S3 S3-Ereignisbenachrichtigungen mit dem AWS CLI

Im Folgenden finden Sie ein Beispiel für einen Amazon S3 AWS CLI S3-Aufruf zum Erstellen von SQS-Warteschlangen und zum Einrichten von Ereignisbenachrichtigungen im Amazon S3 S3-Ziel-Bucket. 


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:

Einrichten eines Crawlers für Amazon S3-Ereignisbenachrichtigungen mit der Konsole (Data Catalog-Ziel)

Wenn Sie ein Catalog Data-Ziel haben, richten Sie einen Crawler für Amazon S3-Ereignisbenachrichtigungen mit der AWS Glue-Konsole ein:

  1. Legen Sie Ihre Crawler-Eigenschaften fest. Weitere Informationen finden Sie unter Festlegen von Crawler-Konfigurationsoptionen auf der AWS Glue-Konsole.

  2. Im Abschnitt Data source configuration (Datenquellenkonfiguration) werden Sie Folgendes gefragt: Sind Ihre Daten bereits AWS Glue-Tabellen zugeordnet?

    Wählen Sie Ja, um vorhandene Tabellen aus Ihrem Data Catalog als Datenquelle auszuwählen.

  3. Wählen Sie im Abschnitt Glue-Tabellen die Option Tabellen hinzufügen.

  4. Konfigurieren Sie im Modal Tabelle hinzufügen die Datenbank und die Tabellen:

    • Network connection (Netzwerkverbindung) (Optional): Wählen Sie Add new connection (Neue Verbindung hinzufügen).

    • Datenbank: Wählen Sie eine Datenbank im Data Catalog.

    • Tabellen: Wählen Sie eine oder mehrere Tabellen aus dieser Datenbank im Data Catalog aus.

    • Subsequent crawler runs (Nachfolgende Crawler-Ausführungen): Wählen Sie Crawl based on events (Crawling basierend auf Ereignissen) aus, um Amazon-S3-Ereignisbenachrichtigungen für Ihren Crawler zu verwenden.

    • SQS ARN hinzufügen: Geben Sie die Datenspeicherparameter einschließlich eines gültigen SQS ARN an. (Beispiel: arn:aws:sqs:region:account:sqs).

    • Dead-Letter SQS ARN hinzufügen (Optional): Geben Sie einen gültigen Amazon Dead-Letter SQS ARN an. (Beispiel: arn:aws:sqs:region:account:deadLetterQueue).

    • Wählen Sie Bestätigen aus.