Настройка потоков автозагрузчика в режиме уведомлений о файлах

На этой странице описывается, как настроить потоки автозагрузчика для использования режима уведомлений файлов для добавочного обнаружения и приема облачных данных.

В режиме уведомлений о файлах Auto Loader автоматически настраивает службу уведомлений и службу очередей, которые подписываются на события с файлами из входного каталога. Вы можете использовать уведомления файлов для масштабирования автозагрузчика для приема миллионов файлов в час. По сравнению с каталожным режимом режим уведомления о файлах быстрее и более масштабируем. Кроме того, вы можете в любое время переключаться между уведомлениями о файлах и списком файлов в каталоге, при этом поддерживая гарантию обработки каждого элемента данных единожды.

Полный справочник по всем параметрам конфигурации автозагрузчика, включая параметры уведомлений о файлах и параметры проверки подлинности в облаке, см. в разделе "Автозагрузчик".

Режим уведомления о файлах с включенными событиями файлов и без них во внешних расположениях

Существует два способа настройки автозагрузчика для использования режима уведомлений файлов:

• (Рекомендуется) События файлов: для всех потоков, обрабатывающих файлы из заданного внешнего расположения, используется одна очередь уведомлений файлов. Этот подход имеет следующие преимущества по сравнению с классическим режимом уведомлений о файлах:
  • Azure Databricks может для вас настраивать подписки и события файлов в учетной записи вашего облачного хранилища без необходимости предоставления дополнительных учетных данных автозагрузчику, используя учетные данные службы или другие параметры аутентификации, специфичные для облака. См. cтатью Настройка событий для файлов во внешнем расположении.
  • У вас меньше политик управляемых удостоверений Azure, создаваемых в вашей учетной записи облачного хранилища.
  • Поскольку вам больше не нужно создавать очередь для каждого потока Auto Loader, проще избежать превышения пределов уведомлений, установленных облачным провайдером, которые перечислены в облачных ресурсах, используемых в классическом режиме уведомления файлов автозагрузчика.
  • Azure Databricks автоматически управляет настройкой требований к ресурсам, поэтому не нужно настраивать такие параметры, как cloudFiles.fetchParallelism.
  • Функциональность очистки означает, что вам не нужно беспокоиться о жизненном цикле уведомлений, созданных в облаке, например при удалении или полном обновлении потока.
• Классический режим уведомлений о файлах: вы управляете очередями уведомлений о файлах для каждого потока автозагрузчика отдельно. Автозагрузчик автоматически настраивает службу уведомлений и службу очередей, подписывающуюся на события файлов из входного каталога. Это классический подход.

Если вы используете автозагрузчик в режиме перечисления каталогов, Databricks рекомендует перенестися в режим уведомлений файлов с событиями файлов. Автозагрузчик с событиями файлов обеспечивает значительные улучшения производительности. Начните с активации событий файлов для внешнего расположения, а затем задайте cloudFiles.useManagedFileEvents в конфигурации потока Auto Loader.

Используйте режим уведомлений о событиях файлов

В этом разделе описывается создание и обновление потоков автозагрузчика для использования событий файлов. Databricks настоятельно рекомендует следующее при использовании режима уведомлений файлов:

• Используйте тома каталога Unity: создайте отдельный внешний том для каждого пути или подкаталога, из которых автозагрузчик загружает данные. Укажите пути тома (например, /Volumes/catalog/schema/volume) для автозагрузчика вместо URL-адресов облачного хранилища (например, s3://bucket/path). Это повышает производительность обнаружения файлов, так как служба событий файлов может ограничить обнаружение файлов только соответствующими объектами, а не итерировать все объекты во внешнем расположении.
• Используйте отдельный том для каждого подпути: если у вас несколько потоков Auto Loader, считывающих из разных подпутей в одном внешнем расположении, создайте выделенный том для каждого подпути вместо совместного использования одного тома. Это позволяет избежать ненужных затрат на обнаружение файлов и помогает предотвратить ограничение скорости.

Перед тем как начать

Для настройки событий файлов требуется:

• Рабочая область Azure Databricks с поддержкой каталога Unity.
• Разрешение на создание учетных данных доступа для хранилища и объектов внешнего расположения в каталоге Unity.

Для потоков автозагрузчика с событиями файлов требуется:

• Выполните вычисления на Databricks Runtime 14.3 LTS или более поздней.

Инструкции по настройке конфигурации

Следующие инструкции применяются к созданию новых потоков автозагрузчика или переносу существующих потоков для использования обновленного режима уведомлений о файлах с событиями файлов:

1. Создайте учетные данные хранилища и внешнее расположение в Unity Catalog, предоставляющее доступ к исходному расположению в облачном хранилище для автозагрузочных потоков.
2. Включите события файлов для внешнего источника. См. cтатью Настройка событий для файлов во внешнем расположении.
3. При создании нового потока Автозагрузчика или изменении существующего для работы с внешним расположением:
   • Если у вас есть потоки автозагрузчика на основе уведомлений, которые используют данные из внешнего источника, выключите их и удалите соответствующие ресурсы уведомлений.
   • Убедитесь, что pathRewrites не установлено (это не является распространённым параметром).
   • Просмотрите список параметров , которые автозагрузчик игнорирует при управлении уведомлениями о файлах с помощью событий файлов. Избегайте их в новых потоках автозагрузчика и удалите их из существующих потоков, которые вы переносите в этот режим.
   • Установите параметр cloudFiles.useManagedFileEvents на true в коде автозагрузчика.

Рассмотрим пример.

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

Если вы используете декларативные конвейеры Lakeflow Spark и уже имеете конвейер с потоковой таблицей, обновите его, чтобы включить этот параметр 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'
                   ...
                   );

Неподдерживаемые параметры автозагрузчика

Следующие параметры автозагрузчика не поддерживаются при использовании потоков событий файлов:

Лучшие практики для событий управляемых файлов см. в Лучших практиках для Auto Loader с событиями файлов.

Ограничения автозагрузчика с событиями файлов

Служба событий файлов оптимизирует обнаружение файлов путем кэширования последних созданных файлов. Если автозагрузчик выполняется редко, этот кэш может истекать, а автозагрузчик возвращается в список каталогов, чтобы обнаружить файлы и обновить кэш. Чтобы избежать этого сценария, вызов автозагрузчика по крайней мере раз в семь дней.

Автозагрузчик с отслеживателем событий файлов следит за FlushWithClose событием для обработки файла. Некоторые операции Azure API могут не генерировать это событие, что может отложить обнаружение файлов. Сведения об этом сценарии см. в разделе "Классические события уведомлений о файлах".

Общий список ограничений для событий файлов см. в разделе "Ограничения событий файлов".

Управление очередями уведомлений файлов для каждого потока автозагрузчика отдельно (классическая модель)

В классическом режиме файлового уведомления автозагрузчик автоматически настраивает выделенную службу уведомлений и очередь для каждого потока. Этот подход требует управления очередями уведомлений для каждого потока и предоставления учетных данных для аутентификации при создании облачных ресурсов. Databricks рекомендует режим уведомлений о файлах для новых рабочих нагрузок.

В классическом режиме уведомлений о файлах Auto Loader автоматически настраивает службу уведомлений и службу очереди для каждого потока, который подписывается на события файлов из входного каталога. Вы управляете очередями уведомлений для каждого потока автозагрузчика отдельно.

Облачные ресурсы, используемые в классическом режиме уведомлений файлов автозагрузчика

Автозагрузчик может автоматически настраивать уведомления о файлах при настройке параметра cloudFiles.useNotificationstrue и предоставления необходимых разрешений для создания облачных ресурсов. Кроме того, может потребоваться предоставить дополнительные параметры для того, чтобы автозагрузчик получил разрешение на создание этих ресурсов.

В следующей таблице перечислены ресурсы, создаваемые автозагрузчиком для каждого поставщика облачных служб.

* Автозагрузчик именует ресурсы с этим префиксом.

**Количество параллельных конвейеров уведомлений о файлах, которые можно запустить.

Если требуется запустить больше потоков автозагрузчика на основе файловых уведомлений, чем разрешено этими ограничениями, можно использовать file events или службы, такие как AWS Lambda, Функции Azure или Google Cloud Functions, чтобы разветвлять уведомления из одной очереди, прослушивающей всю емкость или контейнер, в очереди, относящиеся к определённым каталогам.

Классические события уведомлений о файлах

Amazon S3 предоставляет событие ObjectCreated при загрузке файла в бакет S3, независимо от того, была ли использована однократная или многократная загрузка.

Azure Data Lake Storage предоставляет различные уведомления о событиях для файлов, которые отображаются в контейнере хранилища.

• Автозагрузчик следит за событием FlushWithClose для обработки файла.
• Потоки автозагрузчика поддерживают RenameFile действие обнаружения файлов. RenameFile действия требуют запроса API к системе хранения, чтобы получить размер переименованного файла.
• Потоки Автозагрузчика, созданные с помощью Databricks Runtime 9.0 и более поздних версий, поддерживают действие RenameDirectory для обнаружения файлов. RenameDirectory действия требуют запросов API к системе хранения для перечисления содержимого переименованного каталога.

Google Cloud Storage предоставляет событие OBJECT_FINALIZE при передаче файла, включая перезапись и копирование файлов. Неудачные загрузки не вызывают это событие.

Необходимые разрешения для настройки уведомлений о файлах для Azure Data Lake Storage и Хранилище BLOB-объектов Azure

Необходимо иметь разрешения на чтение для входного каталога. См. раздел Хранилище BLOB-объектов Azure.

Чтобы использовать режим уведомлений файлов, необходимо предоставить учетные данные проверки подлинности для настройки и доступа к службам уведомлений о событиях. Вы можете пройти проверку подлинности с помощью одного из следующих методов:

• В Databricks Runtime 16.1 и более поздних версиях используйте учетные данные службы Azure Databricks: Создайте учетные данные службы с помощью управляемого удостоверения и соединителя доступа Azure Databricks.
• Создайте приложение Microsoft Entra ID (ранее Azure Active Directory) и учетную запись службы в виде идентификатора клиента и секрета клиента.

После того как вы получите учетные данные для аутентификации, назначьте необходимые разрешения, используя один из следующих подходов:

• Встроенные роли Azure

  • Назначьте коннектору доступа следующие роли для учетной записи хранения, в которой находится входной путь:
    • Участник. Эта роль предназначена для настройки ресурсов в учетной записи хранения, таких как очереди и подписки на события.
    • Участник данных очереди хранилища: Эта роль предназначена для выполнения операций с очередями, таких как извлечение и удаление сообщений из них. Эта роль требуется только при использовании служебного принципала без строки подключения.
  • Назначьте соединителю доступа следующую роль связанной группе ресурсов:
    • Участник подписки на события EventGrid: эта роль предназначена для выполнения операций с подписками на события в Сетка событий Azure, таких как создание или перечисление подписок.

• Настраиваемая роль. Если вы обеспокоены предоставлением разрешений, необходимых для предыдущих ролей, можно создать пользовательскую роль со следующими разрешениями. После создания роли назначьте ее соединителю доступа. Дополнительные сведения см. в разделе Назначение ролей Azure в портале 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": []
    }
  ]
  

  

Необходимые разрешения для настройки уведомлений о файлах для Amazon S3

Необходимо иметь разрешения на чтение для входного каталога. Дополнительные сведения см. в сведениях о подключении S3.

Чтобы использовать режим уведомления о файлах, подключите следующий JSON-документ политики к пользователю или роли IAM. Эта роль IAM необходима для создания учетных данных службы, которые "Auto Loader" будет использовать для аутентификации.

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

Замените следующие заполнители значениями для вашей среды:

• <bucket-name>: имя контейнера S3, из которого поток будет считывать файлы, например auto-logs. Можно использовать * в качестве подстановочного знака, например databricks-*-logs. Чтобы узнать базовый контейнер S3 для пути к DBFS, можно перечислить все точки подключения DBFS в записной книжке, выполнив %fs mounts.
• <region>: регион AWS, в котором находится контейнер S3, например, us-west-2. Если вы не хотите указать регион, используйте *.
• <account-number>: номер учетной записи AWS, которой принадлежит контейнер S3, например 123456789012. Если вы не хотите указать номер учетной записи, используйте *.

Строка databricks-auto-ingest-* в спецификации SQS и SNS ARN — это префикс имени, который источник cloudFiles использует при создании служб SQS и SNS. Так как Azure Databricks настраивает службы уведомлений в первоначальном запуске потока, вы можете использовать политику с уменьшенными разрешениями после первоначального запуска (например, для остановки потока и перезапуска).

Ограниченные разрешения после начальной настройки

Вам требуются только разрешения на настройку ресурсов, описанные выше во время начального запуска потока. После первого запуска можно переключиться на следующую политику IAM с ограниченными разрешениями. Однако при снижении разрешений вы не можете запускать новые запросы потоковой передачи, повторно создавать ресурсы после сбоев, таких как случайно удаленная очередь SQS, или использовать API управления облачными ресурсами для перечисления или удаления ресурсов.

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

Необходимые разрешения для настройки уведомлений о файлах для GCS

Чтобы использовать режим уведомлений файлов, необходимо настроить разрешения:

• Убедитесь, что у вас есть list разрешения на контейнер GCS и get разрешения на все объекты. Дополнительные сведения см. в документации Google по разрешениям IAM.

• Добавьте роль Pub/Sub Publisher в учетную запись службы GCS. Это позволит учетной записи публиковать уведомительные сообщения о событиях из контейнеров GCS в службе публикации и подписки Google Cloud.

• Добавьте следующие разрешения в учетную запись службы, используемую для ресурсов Google Cloud Pub или Sub. Вы можете создать пользовательскую роль IAM с этими разрешениями или назначить уже существующие роли GCP для их покрытия. Azure Databricks автоматически создает эту учетную запись службы при создании учетных данных службы. Поддержка учетных данных службы доступна в Databricks Runtime 16.1 и выше.

  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
  

Поиск учетной записи службы GCS

Перейдите к параметрам облачного хранилища > в Google Cloud Console для соответствующего проекта. В разделе "Учетная запись облачной службы хранения" содержится электронная почта учетной записи службы GCS.

Создание настраиваемой роли IAM Google Cloud для режима уведомлений о файлах

Перейдите к ролям IAM и администраторам > в консоли Google Cloud для соответствующего проекта. Затем создайте роль или обновите существующую роль. На странице "Создать роль" выберите "Добавить разрешения". В меню добавьте требуемые разрешения для роли.

Настройка ресурсов уведомлений файлов или управление ими вручную

Привилегированные пользователи могут вручную настраивать ресурсы уведомлений о файлах или управлять ими. Чтобы настроить службы уведомлений файлов через поставщика облачных служб и указать идентификатор очереди, см. уведомление о файле. Чтобы вместо этого использовать API Scala для создания или управления сервисами уведомлений и очереди.

Шаг 1. Создание ResourceManager в AWS, Azure или 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()

язык программирования Scala

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

язык программирования Scala

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

язык программирования Scala

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

Шаг 2. Использование диспетчера ресурсов для настройки, просмотра и удаления служб уведомлений файлов

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

язык программирования Scala

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

Используйте setUpNotificationServices(<resource-suffix>) для создания очереди и подписки с именем <prefix>-<resource-suffix> (префикс зависит от системы хранения, описанной в облачных ресурсах, используемых в классическом режиме уведомлений файла автозагрузчика. Если существует существующий ресурс с тем же именем, Azure Databricks повторно использует существующий ресурс вместо создания нового ресурса. Эта функция возвращает идентификатор очереди, который можно передать в источник cloudFiles, используя идентификатор из Уведомления о файле. Это позволяет исходному пользователю cloudFiles иметь меньше разрешений, чем пользователю, который создает ресурсы.

"path" Укажите параметр newManager только при вызовеsetUpNotificationServices. Он не требуется для listNotificationServices или tearDownNotificationServices. Это тот же path, который вы используете при выполнении потокового запроса.

В следующей матрице указывается, какие методы API поддерживаются в среде выполнения Databricks для каждого типа хранилища:

Очистка ресурсов уведомлений о событиях, созданных автозагрузчиком

Автоматический загрузчик не удаляет ресурсы уведомлений о файлах. Чтобы отключить ресурсы уведомлений о файлах, необходимо использовать диспетчер облачных ресурсов, как показано в предыдущем разделе. Эти ресурсы также можно удалить вручную с помощью пользовательского интерфейса или API поставщика облачных служб.

Устранение распространенных ошибок

В этом разделе описываются распространенные ошибки при использовании автозагрузчика с режимом уведомлений о файлах и их устранении.

Не удалось создать подписку сетки событий

Если при первом запуске автозагрузчика появится следующее сообщение об ошибке, вы не зарегистрировали сетку событий в качестве поставщика ресурсов в подписке Azure.

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

Чтобы зарегистрировать Сетку событий в качестве поставщика ресурсов, сделайте следующее:

1. На портале Azure перейдите к подписке.
2. Выберите поставщики ресурсов в разделе "Параметры".
3. Зарегистрируйте поставщика Microsoft.EventGrid.

Авторизация, необходимая для выполнения операций подписки сетки событий

Если при первом запуске автозагрузчика отображается следующее сообщение об ошибке, убедитесь, что роль участника назначена субъекту-службе для сетки событий и учетной записи хранения.

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

Клиент Event Grid обходит прокси-сервер

В Databricks Runtime 15.2 и более поздних версиях подключения Event Grid в Auto Loader по умолчанию используют параметры прокси-сервера из системных свойств. В Databricks Runtime 13.3 LTS, 14.3 LTS и с 15.0 по 15.2 можно вручную настроить подключения Event Grid для использования прокси-сервера, задав свойство Spark Configspark.databricks.cloudFiles.eventGridClient.useSystemProperties true. См. Установка свойств конфигурации Spark в Azure Databricks.

Слишком много запросов

Если в журналах потоков автозагрузчика отображается следующее сообщение об ошибке, это означает, что потоки превышают ограничение скорости для службы событий файлов Databricks:

com.databricks.sql.util.UnexpectedHttpStatus: Too many requests. Please wait a moment and try again.

Обычно это происходит, когда несколько потоков Auto Loader считывают из разных подпутей в одном внешнем расположении без использования томов каталога Unity. Служба событий файлов должна перебирать все объекты во внешнем хранилище, чтобы найти соответствующие файлы для каждого потока данных, что приводит к чрезмерным вызовам API. Чтобы устранить эту проблему, следуйте рекомендациям, описанным в разделе "Использование режима уведомлений файлов" с событиями файлов.

Дополнительные ресурсы

• Что такое автозагрузчик?
• Настройка потоков автозагрузчика в режиме перечисления каталогов