Konfiguracja strumieni Auto Loader w trybie powiadamiania o plikach

Na tej stronie opisano sposób konfigurowania strumieni Auto Loader do korzystania z trybu powiadamiania plików, aby stopniowo odkrywać i pobierać dane z chmury.

W trybie powiadamiania plików automatyczne moduł ładujący automatycznie konfiguruje usługę powiadomień i usługę kolejki, która subskrybuje zdarzenia plików z katalogu wejściowego. Powiadomienia dotyczące plików umożliwiają skalowanie automatycznego modułu ładującego w celu pozyskiwania milionów plików na godzinę. W porównaniu z trybem wyświetlania listy katalogów tryb powiadomień plików jest szybszy i bardziej skalowalny. Ponadto możesz przełączać się między powiadomieniami o plikach i listami katalogów w dowolnym momencie i nadal utrzymywać dokładnie jednokrotne gwarancje przetwarzania danych.

Aby uzyskać pełną dokumentację wszystkich ustawień konfiguracji automatycznego modułu ładującego, w tym opcji powiadomień dotyczących plików i opcji uwierzytelniania specyficznych dla chmury, zobacz Automatyczne ładowanie.

Tryb powiadomień plików ze zdarzeniami plików włączonymi i wyłączonymi w lokalizacjach zewnętrznych

Istnieją dwa sposoby konfigurowania automatycznego modułu ładującego do korzystania z trybu powiadomień plików:

• (Zalecane) Zdarzenia plików: Należy użyć jednej kolejki powiadomień o plikach dla wszystkich strumieni przetwarzających pliki z danej lokalizacji zewnętrznej. Takie podejście ma następujące zalety w przypadku trybu powiadomień dotyczących plików klasycznych:
  • Azure Databricks może skonfigurować subskrypcje i zdarzenia plików w Twoim koncie magazynu w chmurze bez potrzeby podawania dodatkowych poświadczeń dla Auto Loader, korzystając z poświadczeń usługi lub innych specyficznych dla chmury opcji uwierzytelniania. Zobacz Skonfiguruj zdarzenia plików dla lokalizacji zewnętrznej.
  • Masz mniej zasad zarządzania tożsamościami Azure, do utworzenia w koncie magazynowym w chmurze.
  • Ponieważ nie trzeba już tworzyć kolejki dla każdego strumienia Auto Loader, łatwiej jest uniknąć osiągania limitów powiadomień dostawcy chmury wymienionych w zasobach w chmurze używanych w klasycznym trybie powiadomień Auto Loader.
  • Azure Databricks automatycznie zarządza dostrajaniem wymagań dotyczących zasobów, więc nie musisz dostrajać parametrów, takich jak cloudFiles.fetchParallelism.
  • Funkcja oczyszczania oznacza, że nie musisz się zbytnio martwić o cykl życia powiadomień tworzonych w chmurze, na przykład podczas usuwania strumienia lub jego pełnego odświeżenia.
• Klasyczny tryb powiadamiania o plikach: zarządzasz kolejkami powiadomień plików dla każdego strumienia modułu automatycznego ładowania oddzielnie. Automatyczne ładowanie automatycznie konfiguruje usługę powiadomień i usługę kolejki, która subskrybuje zdarzenia plików z katalogu wejściowego. Jest to podejście klasyczne.

Jeśli używasz Auto Loader w trybie wyświetlania katalogu, usługa Databricks zaleca migrację do trybu powiadamiania o plikach z użyciem zdarzeń plików. Automatyczne ładowanie (Auto Loader) ze zdarzeniami plików zapewnia znaczną poprawę wydajności. Zacznij od włączenia zdarzeń plików dla lokalizacji zewnętrznej, a następnie ustaw cloudFiles.useManagedFileEvents w konfiguracji strumienia Auto Loader.

Używanie trybu powiadomień plików ze zdarzeniami plików

W tej sekcji opisano, jak tworzyć i aktualizować strumienie Auto Loader, aby korzystać ze zdarzeń plików.

Requirements

Konfigurowanie zdarzeń plików wymaga:

• Obszar roboczy Azure Databricks, który jest włączony do katalogu Unity.
• Uprawnienie do tworzenia poświadczeń magazynowych i obiektów lokalizacji zewnętrznej w Unity Catalog.

Strumienie automatycznego modułu ładującego ze zdarzeniami plików wymagają:

• Wykonaj obliczenia w Databricks Runtime 14.3 LTS lub nowszym.

Instrukcje dotyczące konfiguracji

Poniższe instrukcje dotyczą tworzenia nowych strumieni Auto Loader lub migrowania istniejących strumieni w celu korzystania z uaktualnionego trybu powiadomień o zdarzeniach plików:

1. Utwórz poświadczenia magazynowania i lokalizację zewnętrzną w Unity Catalog, które udzielają dostępu do lokalizacji źródłowej w magazynie w chmurze dla strumieni Auto Loader.
2. Włącz zdarzenia plików dla lokalizacji zewnętrznej. Zobacz Skonfiguruj zdarzenia plików dla lokalizacji zewnętrznej.
3. Podczas tworzenia nowego strumienia Auto Loader lub edytowania istniejącego, aby działał z lokalizacją zewnętrzną:
   • Jeśli masz istniejące strumienie Auto Loader oparte na powiadomieniach, które pobierają dane z zewnętrznej lokalizacji, wyłącz je i usuń powiązane zasoby powiadomień.
   • Upewnij się, że pathRewrites nie jest ustawiona (nie jest to typowa opcja).
   • Przejrzyj listę ustawień ignorowanych przez moduł automatycznego ładowania podczas zarządzania powiadomieniami o plikach przy użyciu zdarzeń plików. Unikaj ich w nowych strumieniach Auto Loader i usuń je z istniejących strumieni, które są migrowane do tego trybu.
   • Ustaw opcję cloudFiles.useManagedFileEvents na true w kodzie Auto Loader.

Przykład:

autoLoaderStream = (spark.readStream
  .format("cloudFiles")
  ...
  .options("cloudFiles.useManagedFileEvents", True)
  ...)

Jeśli używasz potoków Lakeflow i masz już potok z tabelą strumieniową, zaktualizuj go, aby uwzględniał opcję useManagedFileEvents:

CREATE OR REFRESH STREAMING LIVE TABLE <table-name>
AS SELECT <select clause expressions>
  FROM STREAM read_files('abfss://path/to/external/location/or/volume',
                   format => '<format>',
                   useManagedFileEvents => 'True'
                   ...
                   );

Nieobsługiwane ustawienia autoloadera

Następujące ustawienia automatycznego modułu ładującego nie są obsługiwane, gdy strumienie używają zdarzeń plików:

Aby zapoznać się z najlepszymi praktykami dotyczącymi zarządzanych zdarzeń plikowych, zobacz Najlepsze praktyki dla Auto Loader ze zdarzeniami plikowymi.

Ograniczenia dotyczące automatycznego modułu ładującego ze zdarzeniami plików

Usługa zdarzeń plików optymalizuje odnajdywanie plików przez buforowanie ostatnio utworzonych plików. Jeśli automatyczne ładowanie jest uruchamiane rzadko, ta pamięć podręczna może wygasnąć, a moduł automatycznego ładowania powróci do listy katalogów w celu odnalezienia plików i zaktualizowania pamięci podręcznej. Aby uniknąć tego scenariusza, należy wywołać moduł automatycznego ładowania co najmniej raz na siedem dni.

Automatyczny załadunek ze zdarzeniami plików nasłuchuje zdarzenia FlushWithClose w celu przetworzenia pliku. Niektóre użycia interfejsu API Azure mogą nie emitować tego zdarzenia, co może opóźnić wyszukiwanie plików. Aby obsłużyć ten scenariusz, zobacz Zdarzenia powiadomień dotyczących plików klasycznych.

Aby uzyskać ogólną listę ograniczeń dotyczących zdarzeń plików, zobacz Ograniczenia zdarzeń plików.

Zarządzanie kolejkami powiadomień plików dla każdego strumienia automatycznego ładowania oddzielnie (wersja klasyczna)

W trybie powiadomień dotyczących plików klasycznych, Auto Loader automatycznie konfiguruje dedykowaną usługę powiadomień i kolejkę dla każdego strumienia. Takie podejście wymaga zarządzania kolejkami powiadomień na strumień i podawania poświadczeń uwierzytelniania na potrzeby tworzenia zasobów w chmurze. Platforma Databricks zaleca tryb powiadomień dotyczących plików dla nowych zadań.

W klasycznym trybie powiadomień dotyczących plików Auto Loader automatycznie konfiguruje usługę powiadomień i usługę kolejkowania dla każdego strumienia subskrybującego zdarzenia plików z katalogu wejściowego. Zarządzasz kolejkami powiadomień dla każdego strumienia Auto Loader osobno.

Zasoby w chmurze używane w klasycznym trybie powiadamiania o plikach automatycznego ładowania

Automatyczne ładowanie może automatycznie konfigurować powiadomienia o plikach po ustawieniu opcji cloudFiles.useNotifications na true i podać niezbędne uprawnienia do tworzenia zasobów w chmurze. Ponadto może być konieczne podanie dodatkowych opcji, aby umożliwić Auto Loaderowi autoryzację do utworzenia tych zasobów.

W poniższej tabeli wymieniono zasoby tworzone przez moduł automatycznego ładowania dla każdego dostawcy usług w chmurze.

* Auto Loader nazywa zasoby używając tego prefiksu.

** Ile współbieżnych potoków powiadomień o plikach można uruchomić

Jeśli musisz uruchomić więcej strumieni automatycznego ładowania opartych na powiadomieniach plików niż dozwalają te limity, możesz użyć zdarzeń plików lub usługi, takiej jak AWS Lambda, Azure Functions lub Google Cloud Functions, aby rozdystrybuować powiadomienia z pojedynczej kolejki, która nasłuchuje całego kontenera lub zasobnika, do kolejek specyficznych dla katalogu.

Klasyczne zdarzenia powiadomień o plikach

Usługa Amazon S3 udostępnia zdarzenie ObjectCreated, gdy plik jest ładowany do zasobnika S3, niezależnie od tego, czy był przesłany za pomocą operacji 'put', czy za pomocą 'multi-part upload'.

Azure Data Lake Storage udostępnia różne powiadomienia o zdarzeniach dla plików pojawiających się w kontenerze przechowywania.

• Moduł automatycznego ładowania nasłuchuje zdarzenia FlushWithClose, aby przetworzyć plik.
• Strumienie Auto Loader obsługują akcję wykrywania plików RenameFile. RenameFile akcje wymagają żądania interfejsu API do systemu pamięci masowej w celu uzyskania rozmiaru zmienionego pliku.
• Strumienie Auto Loader utworzone za pomocą środowiska Databricks Runtime 9.0 i nowszych wspierają akcję RenameDirectory odnajdywania plików. RenameDirectory akcje wymagają, aby żądania interfejsu API do systemu przechowywania wymieniły zawartość zmienionego katalogu.

Usługa Google Cloud Storage zapewnia OBJECT_FINALIZE zdarzenie, gdy plik jest przesyłany, co obejmuje zastępowanie i kopiowanie plików. Nieudane przesyłania nie generują tego zdarzenia.

Wymagane uprawnienia do konfigurowania powiadomień o plikach dla Azure Data Lake Storage i Azure Blob Storage

Musisz mieć uprawnienia do odczytu dla katalogu wejściowego. Zobacz Azure Blob Storage.

Aby użyć trybu powiadomień plików, należy podać poświadczenia uwierzytelniania na potrzeby konfigurowania usług powiadomień zdarzeń i uzyskiwania do nich dostępu. Możesz uwierzytelnić się przy użyciu jednej z następujących metod:

• W środowisku Databricks Runtime 16.1 lub nowszym użyj poświadczeń usługi Azure Databricks: Utwórz poświadczenia usługi przy użyciu tożsamości zarządzanej i łącznika dostępu Azure Databricks.
• Utwórz Microsoft Entra ID (dawniej Azure Active Directory) nazwę główną aplikacji i usługi w postaci identyfikatora klienta i klucza tajnego klienta.

Po uzyskaniu poświadczeń uwierzytelniania przypisz niezbędne uprawnienia przy użyciu jednej z następujących metod:

• Azure wbudowane role:

  • Przypisz łącznikowi dostępu następujące role dla konta magazynu, gdzie znajduje się ścieżka wejściowa:
    • Współautor: Ta rola służy do konfigurowania zasobów na twoim koncie magazynu, takich jak kolejki i subskrypcje zdarzeń.
    • Kontrybutor danych kolejki Storage: rola ta jest dedykowana do wykonywania operacji na kolejce, takich jak pobieranie i usuwanie wiadomości z kolejek. Ta rola jest wymagana tylko w przypadku podania głównego konta usługi bez ciągu połączenia.
  • Przypisz łącznik dostępu następującą rolę do powiązanej grupy zasobów:
    • EventGrid EventSubscription Contributor: Ta rola służy do wykonywania operacji subskrypcji Azure Event Grid (Event Grid), takich jak tworzenie lub wyświetlanie subskrypcji zdarzeń.

• Rola niestandardowa: jeśli chcesz przyznać uprawnienia wymagane dla poprzednich ról, możesz zamiast tego utworzyć rolę niestandardową z następującymi uprawnieniami. Po utworzeniu roli przypisz ją do łącznika dostępu. Aby uzyskać więcej informacji, zobacz Przypisania ról Azure przy użyciu portalu Azure.

  "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": []
    }
  ]
  

  ustawienia uprawnień Auto Loader

Wymagane uprawnienia do konfigurowania powiadomień o plikach dla usługi Amazon S3

Musisz mieć uprawnienia do odczytu dla katalogu wejściowego. Aby uzyskać więcej informacji, zobacz szczegóły połączenia S3.

Aby użyć trybu powiadamiania o plikach, dołącz następujący dokument zasad JSON do swojego użytkownika lub roli IAM. Ta rola IAM jest wymagana do utworzenia poświadczeń usługi dla modułu Auto Loader do celów uwierzytelniania.

{
  "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",
        "sqs:PurgeQueue"
      ],
      "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-*"
      ]
    }
  ]
}

Należy zastąpić następujące symbole zastępcze wartościami dla Twojego środowiska.

• <bucket-name>: nazwa zasobnika S3, z którego strumień będzie odczytywał pliki, na przykład auto-logs. Można użyć * jako symbolu wieloznacznych, na przykład databricks-*-logs. Aby dowiedzieć się podstawowego zasobnika S3 dla ścieżki systemu plików DBFS, możesz wyświetlić listę wszystkich punktów instalacji systemu plików DBFS w notesie, uruchamiając %fs mounts.
• <region>: region platformy AWS, w którym znajduje się zasobnik S3, na przykład us-west-2. Jeśli nie chcesz określać regionu, użyj polecenia *.
• <account-number>: numer konta platformy AWS, który jest właścicielem zasobnika S3, na przykład 123456789012. Jeśli nie chcesz określać numeru konta, użyj polecenia *.

Ciąg databricks-auto-ingest-* w specyfikacji SQS i SNS ARN jest prefiksem nazwy, który źródło cloudFiles używa podczas tworzenia usług SQS i SNS. Ponieważ Azure Databricks konfiguruje usługi powiadomień w początkowym uruchomieniu strumienia, można użyć zasad z ograniczonymi uprawnieniami po początkowym uruchomieniu (na przykład, aby zatrzymać strumień, a następnie uruchomić go ponownie).

Ograniczone uprawnienia po początkowej konfiguracji

Wymagane są tylko uprawnienia konfiguracji zasobów opisane powyżej podczas początkowego uruchomienia strumienia. Po pierwszym uruchomieniu możesz przełączyć się na następującą politykę IAM z ograniczonymi uprawnieniami. Jednak przy ograniczonych uprawnieniach nie można uruchomić nowych zapytań przesyłania strumieniowego, odtworzyć zasoby po awariach, takich jak przypadkowo usunięta kolejka SQS, lub użyć interfejsu API zarządzania zasobami w chmurze do wyświetlania listy lub usuwania zasobów.

{
  "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",
        "sqs:PurgeQueue"
      ],
      "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>:*"
    }
  ]
}

Wymagane uprawnienia do konfigurowania powiadomień o plikach dla usługi GCS

Aby użyć trybu powiadomień plików, należy skonfigurować uprawnienia:

• Step 2: Upewnij się, że masz list oraz get uprawnienia do swojego zasobnika GCS i wszystkich obiektów. Aby uzyskać szczegółowe informacje, zobacz dokumentację firmy Google dotyczącą uprawnień IAM.

• Dodaj rolę Pub/Sub Publisher do konta usługi GCS. Dzięki temu konto może publikować komunikaty powiadomień o zdarzeniach z zasobników GCS do usługi Google Cloud Pub/Sub.

• Dodaj następujące uprawnienia do konta usługi używanego dla zasobów Google Cloud Pub/Sub. Możesz utworzyć rolę niestandardową IAM z tymi uprawnieniami lub przypisać istniejące role GCP , aby je uwzględnić. Azure Databricks automatycznie tworzy to konto usługi podczas tworzenia poświadczenia usługi. Obsługa poświadczeń usługi jest dostępna w środowisku Databricks Runtime 16.1 lub nowszym.

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

Znajdowanie konta usługi GCS

Przejdź do Ustawienia usługi Cloud Storage w Google Cloud Console dla odpowiadającego projektu. Sekcja Konto usługi Google Cloud Storage zawiera adres e-mail konta usługi Google Cloud Storage.

Tworzenie niestandardowej roli IAM w chmurze Google dla trybu powiadomień plików

Przejdź do IAM & Admin Roles w konsoli Google Cloud dla odpowiedniego projektu. Następnie utwórz rolę lub zaktualizuj istniejącą rolę. Na stronie Tworzenie roli wybierz pozycję Dodaj uprawnienia. W menu dodaj odpowiednie uprawnienia do roli.

Ręczne konfigurowanie zasobów powiadomień o plikach lub zarządzanie nimi

Użytkownicy uprzywilejowani mogą ręcznie konfigurować zasoby powiadomień o plikach lub zarządzać nimi. Aby skonfigurować usługi powiadomień plików za pośrednictwem dostawcy usług w chmurze i określić identyfikator kolejki, zobacz Powiadomienie o pliku. Aby użyć interfejsów API języka Scala do tworzenia powiadomień i usług kolejkowania lub zarządzania nimi zamiast tego:

Krok 1. Tworzenie menedżera zasobów na platformie AWS, Azure lub Google Cloud

Python

# Create a ResourceManager in AWS

# Using a Databricks service credential
manager = spark._jvm.com.databricks.sql.CloudFilesAWSResourceManager \
  .newManager() \
  .option("cloudFiles.region", <region>) \
  .option("path", <path-to-specific-bucket-and-folder>) \
  .option("databricks.serviceCredential", <service-credential-name>) \
  .create()

# Using an AWS access key and secret key
manager = spark._jvm.com.databricks.sql.CloudFilesAWSResourceManager \
  .newManager() \
  .option("cloudFiles.region", <region>) \
  .option("cloudFiles.awsAccessKey", <aws-access-key>) \
  .option("cloudFiles.awsSecretKey", <aws-secret-key>) \
  .option("cloudFiles.roleArn", <role-arn>) \
  .option("cloudFiles.roleExternalId", <role-external-id>) \
  .option("cloudFiles.roleSessionName", <role-session-name>) \
  .option("cloudFiles.stsEndpoint", <sts-endpoint>) \
  .option("path", <path-to-specific-bucket-and-folder>) \
  .create()

Skala

// Create a ResourceManager in AWS

import com.databricks.sql.CloudFilesAWSResourceManager

// Using a Databricks service credential
val manager = CloudFilesAWSResourceManager
    .newManager
    .option("cloudFiles.region", <region>) // optional, will use the region of the EC2 instances by default
    .option("databricks.serviceCredential", <service-credential-name>)
    .option("path", <path-to-specific-bucket-and-folder>) // required only for setUpNotificationServices
    .create()

// Using AWS access key and secret key
val manager = CloudFilesAWSResourceManager
    .newManager
    .option("cloudFiles.region", <region>)
    .option("cloudFiles.awsAccessKey", <aws-access-key>)
    .option("cloudFiles.awsSecretKey", <aws-secret-key>)
    .option("cloudFiles.roleArn", <role-arn>)
    .option("cloudFiles.roleExternalId", <role-external-id>)
    .option("cloudFiles.roleSessionName", <role-session-name>)
    .option("cloudFiles.stsEndpoint", <sts-endpoint>)
    .option("path", <path-to-specific-bucket-and-folder>) // required only for setUpNotificationServices
    .create()

Python

# Create a ResourceManager in Azure

# Using a Databricks service credential
manager = spark._jvm.com.databricks.sql.CloudFilesAzureResourceManager \
  .newManager() \
  .option("cloudFiles.resourceGroup", <resource-group>) \
  .option("cloudFiles.subscriptionId", <subscription-id>) \
  .option("databricks.serviceCredential", <service-credential-name>) \
  .option("path", <path-to-specific-container-and-folder>) \
  .create()

# Using an Azure service principal
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()

Skala

// Create a ResourceManager in Azure

import com.databricks.sql.CloudFilesAzureResourceManager

// Using a Databricks service credential
val manager = CloudFilesAzureResourceManager
  .newManager
  .option("cloudFiles.resourceGroup", <resource-group>)
  .option("cloudFiles.subscriptionId", <subscription-id>)
  .option("databricks.serviceCredential", <service-credential-name>)
  .option("path", <path-to-specific-container-and-folder>) // required only for setUpNotificationServices
  .create()

// Using an Azure service principal
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()

Python

# Create a ResourceManager in GCP

# Using a Databricks service credential
manager = spark._jvm.com.databricks.sql.CloudFilesGCPResourceManager \
  .newManager() \
  .option("cloudFiles.projectId", <project-id>) \
  .option("databricks.serviceCredential", <service-credential-name>) \
  .option("path", <path-to-specific-bucket-and-folder>) \
  .create()

# Using a Google service account
manager = spark._jvm.com.databricks.sql.CloudFilesGCPResourceManager \
  .newManager() \
  .option("cloudFiles.projectId", <project-id>) \
  .option("cloudFiles.client", <client-id>) \
  .option("cloudFiles.clientEmail", <client-email>) \
  .option("cloudFiles.privateKey", <private-key>) \
  .option("cloudFiles.privateKeyId", <private-key-id>) \
  .option("path", <path-to-specific-bucket-and-folder>) \
  .create()

Skala

// Create a ResourceManager in GCP

import com.databricks.sql.CloudFilesGCPResourceManager

// Using a Databricks service credential
val manager = CloudFilesGCPResourceManager
    .newManager
    .option("cloudFiles.projectId", <project-id>)
    .option("databricks.serviceCredential", <service-credential-name>)
    .option("path", <path-to-specific-bucket-and-folder>) // Required only for setUpNotificationServices.
    .create()

// Using a Google service account
val manager = CloudFilesGCPResourceManager
    .newManager
    .option("cloudFiles.projectId", <project-id>)
    .option("cloudFiles.client", <client-id>)
    .option("cloudFiles.clientEmail", <client-email>)
    .option("cloudFiles.privateKey", <private-key>)
    .option("cloudFiles.privateKeyId", <private-key-id>)
    .option("path", <path-to-specific-bucket-and-folder>) // Required only for setUpNotificationServices.
    .create()

Krok 2. Konfigurowanie, wyświetlanie i odłączenie usług powiadomień plików za pomocą menedżera zasobów

Python

# 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>)

Skala

// 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>)

Użyj setUpNotificationServices(<resource-suffix>) do utworzenia kolejki i subskrypcji o nazwie <prefix>-<resource-suffix> (prefiks zależy od systemu magazynu podsumowanego w zasobach chmury używanych w klasycznym trybie powiadamiania plikowego Auto Loader). Jeśli istnieje zasób o tej samej nazwie, Azure Databricks ponownie użyje istniejącego zasobu zamiast utworzyć nowy. Ta funkcja zwraca identyfikator kolejki, który można przekazać do źródła cloudFiles za pomocą identyfikatora z Powiadomienia o pliku. Dzięki cloudFiles temu użytkownik źródłowy może mieć mniej uprawnień niż użytkownik tworzący zasoby.

"path" Podaj opcję tylko newManager w przypadku wywołania metody setUpNotificationServices. Nie jest to wymagane dla listNotificationServices ani tearDownNotificationServices. Jest to ten sam path, którego używasz, gdy uruchamiasz zapytanie przesyłania strumieniowego.

Poniższa macierz wskazuje, które metody interfejsu API są obsługiwane w środowisku Databricks Runtime dla każdego typu magazynu:

Czyszczenie zasobów powiadomień o zdarzeniach utworzonych przez moduł automatycznego ładowania

Automatyczne ładowanie nie powoduje automatycznego usuwania zasobów powiadomień o plikach. Aby usunąć zasoby powiadomień o plikach, należy użyć menedżera zasobów w chmurze, jak pokazano w poprzedniej sekcji. Te zasoby można również usunąć ręcznie przy użyciu interfejsu użytkownika lub interfejsów API dostawcy usług w chmurze.

Rozwiązywanie typowych błędów

W tej sekcji opisano typowe błędy występujące podczas korzystania z automatycznego modułu ładującego z trybem powiadamiania plików i sposobu ich rozwiązywania.

Nie można utworzyć subskrypcji usługi Event Grid

Jeśli podczas pierwszego uruchamiania automatycznego modułu ładującego zostanie wyświetlony następujący komunikat o błędzie, usługa Event Grid nie została zarejestrowana jako dostawca zasobów w subskrypcji Azure.

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

Aby zarejestrować usługę Event Grid jako dostawcę zasobów, wykonaj następujące czynności:

1. W portalu Azure przejdź do subskrypcji.
2. Wybierz pozycję Dostawcy zasobów w sekcji Ustawienia.
3. Zarejestruj dostawcę Microsoft.EventGrid.

Autoryzacja wymagana do wykonywania operacji subskrypcji usługi Event Grid

Jeśli podczas pierwszego uruchamiania Auto Loader'a wyświetli się następujący komunikat o błędzie, upewnij się, że rola Współautora jest przypisana do głównej usługi dla Event Grid i konta magazynu.

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

Klient usługi Event Grid pomija serwer proxy

W środowisku Databricks Runtime 15.2 lub nowszym połączenia usługi Event Grid w usłudze Auto Loader domyślnie używają ustawień serwera proxy z właściwości systemowych. W Databricks Runtime 13.3 LTS, 14.3 LTS i od 15.0 do 15.2 można ręcznie skonfigurować połączenia Event Grid, aby korzystać z serwera proxy, ustawiając właściwość Spark Configspark.databricks.cloudFiles.eventGridClient.useSystemProperties true. Zobacz Ustawianie właściwości konfiguracji Spark na Azure Databricks.

Dodatkowe zasoby

• Co to jest moduł automatycznego ładowania?
• Konfigurować strumienie Auto Loader w trybie listy katalogów