Was ist der Dateibenachrichtrigungsmodus von Autoloader?

Im Dateibenachrichtigungsmodus richtet Auto Loader automatisch einen Benachrichtigungsdienst und einen Warteschlangendienst ein, der Dateiereignisse aus dem Eingabeverzeichnis abonniert. Sie können Dateibenachrichtigungen verwenden, um den Autoloader so zu skalieren, dass pro Stunde Millionen von Dateien erfasst werden. Im Vergleich zum Verzeichnisauflistungsmodus ist der Dateibenachrichtigungsmodus leistungsfähiger und skalierbarer für große Eingabeverzeichnisse oder eine große Anzahl von Dateien, erfordert jedoch zusätzliche Cloudberechtigungen.

Sie können jederzeit zwischen Dateibenachrichtigungen und Verzeichnisauflistung wechseln und trotzdem die Garantien für die einmalige Datenverarbeitung einhalten.

Der Dateibenachrichtigungsmodus wird nur für Einzelbenutzer-Compute unterstützt.

Cloudressourcen, die im Autoloader-Dateibenachrichtigungsmodus verwendet werden

Der Autoloader kann Dateibenachrichtigungen automatisch für Sie einrichten, wenn Sie die Option cloudFiles.useNotifications auf true festlegen und die erforderlichen Berechtigungen zum Erstellen von Cloudressourcen bereitstellen. Darüber hinaus müssen Sie möglicherweise zusätzlichen Optionen bereitstellen, um Autoloader die Berechtigung zum Erstellen dieser Ressourcen zu gewähren.

In der folgenden Tabelle wird zusammengefasst, welche Ressourcen vom Autoloader erstellt werden.

• Auto Loader benennt die Ressourcen mit diesem Präfix.

** Wie viele gleichzeitige Dateibenachrichtigungspipelines können gestartet werden

Wenn Sie mehr als die begrenzte Anzahl von Dateibenachrichtigungspipelines für ein bestimmtes Speicherkonto ausführen müssen, haben Sie folgende Möglichkeiten:

• Nutzen Sie einen Service wie AWS Lambda, Azure Functions oder Google Cloud Functions, um Benachrichtigungen von einer einzigen Warteschlange, die einen gesamten Container oder Bucket abhört, auf verzeichnisspezifische Warteschlangen aufzuteilen.

Dateibenachrichtigungsereignisse

AWS S3 stellt ein ObjectCreated-Ereignis bereit, wenn eine Datei in einen S3-Bucket hochgeladen wird, unabhängig davon, ob sie durch einen Put- oder einen mehrteiligen Upload hochgeladen wurde.

ADLS Gen2 stellt verschiedene Ereignisbenachrichtigungen für Dateien bereit, die in Ihrem Gen2-Container angezeigt werden.

• Der Autoloader lauscht auf das FlushWithClose-Ereignis für die Verarbeitung einer Datei.
• Auto Loader Streams unterstützen die RenameFile Aktion zum Ermitteln von Dateien. RenameFile-Aktionen erfordern eine API-Anforderung an das Speichersystem, um die Größe der umbenannten Datei abzurufen.
• Autoloader-Streams, die mit Databricks Runtime 9.0 und höher erstellt wurden, unterstützen die RenameDirectory-Aktion zum Ermitteln von Dateien. RenameDirectory-Aktionen erfordern API-Anforderungen an das Speichersystem, um den Inhalt des umbenannten Verzeichnisses aufzulisten.

Google Cloud Storage stellt ein OBJECT_FINALIZE-Ereignis bereit, wenn eine Datei hochgeladen wird, einschließlich Überschreibungen und Dateikopien. Bei fehlgeschlagenen Uploads wird dieses Ereignis nicht generiert.

Erforderliche Berechtigungen zum Konfigurieren der Dateibenachrichtigung für ADLS Gen2 und Azure Blob Storage

Sie müssen über Leseberechtigungen für das Eingabeverzeichnis verfügen. Weitere Informationen finden Sie unter Azure Blob Storage.

Zur Verwendung des Dateibenachrichtigungsmodus müssen Sie Anmeldeinformationen für die Authentifizierung angeben, um die Ereignisbenachrichtigungsdienste einrichten und darauf zugreifen zu können. Sie benötigen nur einen Dienstprinzipal für die Authentifizierung.

• Dienstprinzipal – Verwenden von integrierten Azure-Rollen

  Erstellen Sie eine Microsoft Entra ID-Anwendung (ehemals Azure Active Directory) und einen Dienstprinzipal in Form einer Client-ID und eines geheimen Clientschlüssels.

  Weisen Sie dieser App die folgenden Rollen für das Speicherkonto zu, in dem sich der Eingabepfad befindet:

  • Mitwirkender: Diese Rolle dient zum Einrichten von Ressourcen in Ihrem Speicherkonto, z. B. Warteschlangen und Ereignisabonnements.
  • Mitwirkender an Storage-Warteschlangendaten: Diese Rolle dient zum Ausführen von Warteschlangenvorgänge, z. B. Abrufen und Löschen von Nachrichten aus den Warteschlangen. Diese Rolle ist nur erforderlich, wenn Sie einen Dienstprinzipal ohne Verbindungszeichenfolge bereitstellen.

  Weisen Sie dieser App die folgende Rolle für die zugehörige Ressourcengruppe zu:

  • EventGrid EventSubscription-Mitwirkender: Diese Rolle dient zum Ausführen von Event Grid-Abonnementvorgängen, z. B. Erstellen oder Auflisten von Ereignisabonnements.

  Weitere Informationen finden Sie unter Hinzufügen oder Entfernen von Azure-Rollenzuweisungen über das Azure-Portal.

• Dienstprinzipal – Verwenden von benutzerdefinierter Rolle

  Wenn Sie Bedenken wegen der übermäßigen Berechtigungen haben, die für die vorherigen Rollen erforderlich sind, können Sie eine benutzerdefinierte Rolle mit mindestens den folgenden Berechtigungen erstellen, die unten im JSON-Format der Azure-Rolle aufgeführt sind:

  "permissions": [
    {
      "actions": [
        "Microsoft.EventGrid/eventSubscriptions/write",
        "Microsoft.EventGrid/eventSubscriptions/read",
        "Microsoft.EventGrid/eventSubscriptions/delete",
        "Microsoft.EventGrid/locations/eventSubscriptions/read",
        "Microsoft.Storage/storageAccounts/read",
        "Microsoft.Storage/storageAccounts/write",
        "Microsoft.Storage/storageAccounts/queueServices/read",
        "Microsoft.Storage/storageAccounts/queueServices/write",
        "Microsoft.Storage/storageAccounts/queueServices/queues/write",
        "Microsoft.Storage/storageAccounts/queueServices/queues/read",
        "Microsoft.Storage/storageAccounts/queueServices/queues/delete"
    ],
      "notActions": [],
      "dataActions": [
        "Microsoft.Storage/storageAccounts/queueServices/queues/messages/delete",
        "Microsoft.Storage/storageAccounts/queueServices/queues/messages/read",
        "Microsoft.Storage/storageAccounts/queueServices/queues/messages/write",
        "Microsoft.Storage/storageAccounts/queueServices/queues/messages/process/action"
      ],
      "notDataActions": []
    }
  ]
  

  Anschließend können Sie diese benutzerdefinierte Rolle Ihrer App zuweisen.

  Weitere Informationen finden Sie unter Hinzufügen oder Entfernen von Azure-Rollenzuweisungen über das Azure-Portal.

Problembehandlung bei häufige Fehlern

Fehler:

java.lang.RuntimeException: Failed to create event grid subscription.

Wenn beim ersten Ausführen des Autoloaders diese Fehlermeldung angezeigt wird, ist der Event Grid nicht als Ressourcenanbieter in Ihrem Azure-Abonnement registriert. Zum Registrieren über das Azure-Portal gehen Sie wie folgt vor:

1. Wechseln Sie zu Ihrem Abonnement.
2. Klicken Sie im Abschnitt „Einstellungen“ auf Ressourcenanbieter.
3. Registrieren Sie den Anbieter Microsoft.EventGrid.

Fehler:

403 Forbidden ... does not have authorization to perform action 'Microsoft.EventGrid/eventSubscriptions/[read|write]' over scope ...

Wenn beim ersten Ausführen des Autoloaders diese Fehlermeldung angezeigt wird, stellen Sie sicher, dass Sie dem Dienstprinzipal für Event Grid sowie Ihrem Speicherkonto die Rolle Mitwirkender zugewiesen haben.

Erforderliche Berechtigungen zum Konfigurieren der Dateibenachrichtigung für AWS S3

Sie müssen über Leseberechtigungen für das Eingabeverzeichnis verfügen. Weitere Informationen finden Sie unter S3-Verbindungsdetails.

Um den Dateibenachrichtigungsmodus zu verwenden, fügen Sie das folgende JSON-Richtliniendokument an Ihren IAM-Benutzer oder Ihre Rolle an.

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "DatabricksAutoLoaderSetup",
      "Effect": "Allow",
      "Action": [
        "s3:GetBucketNotification",
        "s3:PutBucketNotification",
        "sns:ListSubscriptionsByTopic",
        "sns:GetTopicAttributes",
        "sns:SetTopicAttributes",
        "sns:CreateTopic",
        "sns:TagResource",
        "sns:Publish",
        "sns:Subscribe",
        "sqs:CreateQueue",
        "sqs:DeleteMessage",
        "sqs:ReceiveMessage",
        "sqs:SendMessage",
        "sqs:GetQueueUrl",
        "sqs:GetQueueAttributes",
        "sqs:SetQueueAttributes",
        "sqs:TagQueue",
        "sqs:ChangeMessageVisibility"
      ],
      "Resource": [
        "arn:aws:s3:::<bucket-name>",
        "arn:aws:sqs:<region>:<account-number>:databricks-auto-ingest-*",
        "arn:aws:sns:<region>:<account-number>:databricks-auto-ingest-*"
      ]
    },
    {
      "Sid": "DatabricksAutoLoaderList",
      "Effect": "Allow",
      "Action": [
        "sqs:ListQueues",
        "sqs:ListQueueTags",
        "sns:ListTopics"
      ],
      "Resource": "*"
    },
    {
      "Sid": "DatabricksAutoLoaderTeardown",
      "Effect": "Allow",
      "Action": [
        "sns:Unsubscribe",
        "sns:DeleteTopic",
        "sqs:DeleteQueue"
      ],
      "Resource": [
        "arn:aws:sqs:<region>:<account-number>:databricks-auto-ingest-*",
        "arn:aws:sns:<region>:<account-number>:databricks-auto-ingest-*"
      ]
    }
  ]
}

Hierbei gilt:

• <bucket-name>: Der Name des S3-Buckets, in dem der Stream Dateien liest, z. B. auto-logs. Sie können * als Platzhalter verwenden, z. B. databricks-*-logs. Um den zugrunde liegenden S3-Bucket für Ihren DBFS-Pfad zu ermitteln, können Sie durch Ausführung von %fs mounts alle DBFS-Bereitstellungspunkte in einem Notebook auflisten.
• <region>: Die AWS-Region, in der sich der S3-Bucket befindet, z. B. us-west-2. Wenn Sie die Region nicht angeben möchten, verwenden Sie *.
• <account-number>: Die Nummer des AWS-Kontos, das als Besitzer des S3-Buckets fungiert, z. B. 123456789012. Wenn Sie die Kontonummer nicht angeben möchten, verwenden Sie *.

Die Zeichenfolge databricks-auto-ingest-* in der SQS- und SNS-ARN-Spezifikation ist das Namenspräfix, das die cloudFiles-Quelle beim Erstellen von SQS- und SNS-Diensten verwendet. Da Azure Databricks die Benachrichtigungsdienste bei der ersten Ausführung des Streams einrichtet, können Sie nach der ersten Ausführung eine Richtlinie mit eingeschränkten Berechtigungen verwenden (halten Sie den Stream beispielsweise an, und starten Sie ihn dann neu).

• s3:ListBucket
• s3:GetObject

Eingeschränkte Berechtigungen nach der Ersteinrichtung

Die oben beschriebenen Berechtigungen zum Einrichten von Ressourcen werden nur bei der ersten Ausführung des Streams benötigt. Nach der ersten Ausführung können Sie zur folgenden IAM-Richtlinie mit reduzierten Berechtigungen wechseln.

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "DatabricksAutoLoaderUse",
      "Effect": "Allow",
      "Action": [
       "s3:GetBucketNotification",
       "sns:ListSubscriptionsByTopic",
       "sns:GetTopicAttributes",
       "sns:TagResource",
       "sns:Publish",
       "sqs:DeleteMessage",
       "sqs:ReceiveMessage",
       "sqs:SendMessage",
       "sqs:GetQueueUrl",
       "sqs:GetQueueAttributes",
       "sqs:TagQueue",
       "sqs:ChangeMessageVisibility"
      ],
      "Resource": [
       "arn:aws:sqs:<region>:<account-number>:<queue-name>",
       "arn:aws:sns:<region>:<account-number>:<topic-name>",
       "arn:aws:s3:::<bucket-name>"
      ]
    },
    {
      "Effect": "Allow",
      "Action": [
       "s3:GetBucketLocation",
       "s3:ListBucket"
      ],
      "Resource": [
       "arn:aws:s3:::<bucket-name>"
      ]
    },
    {
      "Effect": "Allow",
      "Action": [
       "s3:PutObject",
       "s3:PutObjectAcl",
       "s3:GetObject",
       "s3:DeleteObject"
      ],
      "Resource": [
       "arn:aws:s3:::<bucket-name>/*"
      ]
    },
    {
      "Sid": "DatabricksAutoLoaderListTopics",
      "Effect": "Allow",
      "Action": [
       "sqs:ListQueues",
       "sqs:ListQueueTags",
       "sns:ListTopics"
      ],
      "Resource": "arn:aws:sns:<region>:<account-number>:*"
    }
  ]
}

Erforderliche Berechtigungen zum Konfigurieren der Dateibenachrichtigung für GCS

Sie müssen über list- und get-Berechtigungen für Ihren GCS-Bucket und für alle Objekte verfügen. Weitere Informationen finden Sie in der Google-Dokumentation zu IAM-Berechtigungen.

Um den Dateibenachrichtigungsmodus zu verwenden, müssen Sie für das GCS-Dienstkonto und für das Konto Berechtigungen hinzufügen, das für den Zugriff auf die Google Cloud Pub/Sub-Ressourcen verwendet wird.

Fügen Sie die Pub/Sub Publisher-Rolle dem GCS-Dienstkonto hinzu. Dadurch kann das Konto Ereignisbenachrichtigungen aus Ihren GCS-Buckets in Google Cloud Pub/Sub veröffentlichen.

Für das Dienstkonto, das für die Google Cloud Pub/Sub-Ressourcen verwendet wird, müssen Sie die folgenden Berechtigungen hinzufügen:

pubsub.subscriptions.consume
pubsub.subscriptions.create
pubsub.subscriptions.delete
pubsub.subscriptions.get
pubsub.subscriptions.list
pubsub.subscriptions.update
pubsub.topics.attachSubscription
pubsub.topics.create
pubsub.topics.delete
pubsub.topics.get
pubsub.topics.list
pubsub.topics.update

Hierzu können Sie entweder eine benutzerdefinierte IAM-Rolle mit diesen Berechtigungen erstellen oder bereits vorhandene GCP-Rollen zuweisen, um diese Berechtigungen abzudecken.

Suchen des GCS-Dienstkontos

Navigieren Sie in der Google Cloud Console für das entsprechende Projekt zu Cloud Storage > Settings. Der Abschnitt „Cloud Storage Service Account“ enthält die E-Mail des GCS-Dienstkontos.

Erstellen einer benutzerdefinierten Google Cloud-IAM-Rolle für den Dateibenachrichtigungsmodus

Navigieren Sie in der Google Cloud Console für das entsprechende Projekt zu IAM & Admin > Roles. Erstellen Sie dann entweder oben eine Rolle, oder aktualisieren Sie eine vorhandene Rolle. Klicken Sie auf dem Bildschirm für die Rollenerstellung oder -bearbeitung auf Add Permissions. Es erscheint ein Menü, in dem Sie der Rolle die gewünschten Berechtigungen hinzufügen können.

Manuelles Konfigurieren oder Verwalten von Dateibenachrichtigungsressourcen.

Privilegierte Benutzer können Dateibenachrichtigungsressourcen manuell konfigurieren oder verwalten.

• Richten Sie die Dateibenachrichtigungsdienste manuell über den Cloudanbieter ein, und geben Sie den Warteschlangenbezeichner manuell an. Weitere Informationen finden Sie unter Dateibenachrichtigungsoptionen.
• Verwenden Sie Scala-APIs, um die Benachrichtigungs- und Warteschlangendienste zu erstellen oder zu verwalten, wie im folgenden Beispiel gezeigt:

Python

# Databricks notebook source
# MAGIC %md ## Python bindings for CloudFiles Resource Managers for all 3 clouds

# COMMAND ----------

#####################################
## Creating a ResourceManager in AWS
#####################################

manager = spark._jvm.com.databricks.sql.CloudFilesAWSResourceManager \
  .newManager() \
  .option("cloudFiles.region", <region>) \
  .option("path", <path-to-specific-bucket-and-folder>) \
  .create()

#######################################
## Creating a ResourceManager in Azure
#######################################

manager = spark._jvm.com.databricks.sql.CloudFilesAzureResourceManager \
  .newManager() \
  .option("cloudFiles.connectionString", <connection-string>) \
  .option("cloudFiles.resourceGroup", <resource-group>) \
  .option("cloudFiles.subscriptionId", <subscription-id>) \
  .option("cloudFiles.tenantId", <tenant-id>) \
  .option("cloudFiles.clientId", <service-principal-client-id>) \
  .option("cloudFiles.clientSecret", <service-principal-client-secret>) \
  .option("path", <path-to-specific-container-and-folder>) \
  .create()

#######################################
## Creating a ResourceManager in GCP
#######################################
manager = spark._jvm.com.databricks.sql.CloudFilesGCPResourceManager \
  .newManager() \
  .option("path", <path-to-specific-bucket-and-folder>) \
  .create()

# Set up a queue and a topic subscribed to the path provided in the manager.
manager.setUpNotificationServices(<resource-suffix>)

# List notification services created by <AL>
from pyspark.sql import DataFrame
df = DataFrame(manager.listNotificationServices(), spark)

# Tear down the notification services created for a specific stream ID.
# Stream ID is a GUID string that you can find in the list result above.
manager.tearDownNotificationServices(<stream-id>)

Scala

/////////////////////////////////////
// Creating a ResourceManager in AWS
/////////////////////////////////////

import com.databricks.sql.CloudFilesAWSResourceManager
val manager = CloudFilesAWSResourceManager
    .newManager
    .option("cloudFiles.region", <region>) // optional, will use the region of the EC2 instances by default
    .option("path", <path-to-specific-bucket-and-folder>) // required only for setUpNotificationServices
    .create()

///////////////////////////////////////
// Creating a ResourceManager in Azure
///////////////////////////////////////

import com.databricks.sql.CloudFilesAzureResourceManager
val manager = CloudFilesAzureResourceManager
  .newManager
  .option("cloudFiles.connectionString", <connection-string>)
  .option("cloudFiles.resourceGroup", <resource-group>)
  .option("cloudFiles.subscriptionId", <subscription-id>)
  .option("cloudFiles.tenantId", <tenant-id>)
  .option("cloudFiles.clientId", <service-principal-client-id>)
  .option("cloudFiles.clientSecret", <service-principal-client-secret>)
  .option("path", <path-to-specific-container-and-folder>) // required only for setUpNotificationServices
  .create()

///////////////////////////////////////
// Creating a ResourceManager in GCP
///////////////////////////////////////

import com.databricks.sql.CloudFilesGCPResourceManager
val manager = CloudFilesGCPResourceManager
    .newManager
    .option("path", <path-to-specific-bucket-and-folder>) // Required only for setUpNotificationServices.
    .create()

// Set up a queue and a topic subscribed to the path provided in the manager.
manager.setUpNotificationServices(<resource-suffix>)

// List notification services created by <AL>
val df = manager.listNotificationServices()

// Tear down the notification services created for a specific stream ID.
// Stream ID is a GUID string that you can find in the list result above.
manager.tearDownNotificationServices(<stream-id>)

Verwenden Sie setUpNotificationServices(<resource-suffix>), um eine Warteschlange und ein Abonnement mit dem Namen <prefix>-<resource-suffix> zu erstellen (das Präfix hängt vom Speichersystem ab, das unter Im Autoloader-Dateibenachrichtigungsmodus verwendete Cloudressourcen zusammengefasst ist). Wenn eine vorhandene Ressource mit demselben Namen vorhanden ist, verwendet Azure Databricks die vorhandene Ressource erneut, anstatt eine neue zu erstellen. Diese Funktion gibt einen Warteschlangenbezeichner zurück, den Sie mithilfe des Bezeichners in Dateibenachrichtigungsoptionen an die cloudFiles-Quelle übergeben können. Dadurch hat der Benutzer der cloudFiles-Quelle weniger Berechtigungen als der Benutzer, der die Ressourcen erstellt.

Geben Sie die Option "path" für newManager nur an, wenn Sie setUpNotificationServices aufrufen. Sie wird für listNotificationServices oder tearDownNotificationServices nicht benötigt. Dies ist derselbe path, den Sie beim Ausführen einer Streamingabfrage verwenden.

Die folgende Matrix gibt an, welche API-Methoden in welcher Databricks Runtime für jeden Speichertyp unterstützt wird: