Configurar fluxos do Auto Loader no modo de notificação de arquivo

Esta página descreve como configurar fluxos do Auto Loader para usar o modo de notificação de arquivo para descobrir e ingerir dados na nuvem de forma incremental.

No modo de notificação de ficheiros, o Carregador Automático configura automaticamente um serviço de notificação e um serviço de fila que subscreve eventos de ficheiros a partir do diretório de entrada. Você pode usar notificações de arquivo para dimensionar o Auto Loader para ingerir milhões de arquivos por hora. Quando comparado ao modo de listagem de diretórios, o modo de notificação de arquivos é mais eficiente e escalável.

Você pode alternar entre notificações de arquivos e listas de diretórios a qualquer momento e ainda manter garantias de processamento de dados exatamente uma vez.

Modo de notificação de ficheiros com e sem eventos de ficheiros ativados em locais externos

Há duas maneiras de configurar o Auto Loader para usar o modo de notificação de arquivo:

• (Recomendado) Eventos de ficheiro: utiliza-se uma fila de notificação de ficheiro único para todos os fluxos que processam ficheiros a partir de uma determinada localização externa.

  Essa abordagem tem as seguintes vantagens em relação ao modo de notificação de arquivo herdado:

  • O Azure Databricks pode configurar assinaturas e eventos de arquivo em sua conta de armazenamento em nuvem para você sem exigir que você forneça credenciais adicionais ao Auto Loader usando uma credencial de serviço ou outras opções de autenticação específicas da nuvem. Veja (Recomendado) Ativar eventos de ficheiro para um local externo.
  • Você tem menos políticas de identidade gerenciada do Azure para criar em sua conta de armazenamento em nuvem.
  • Como você não precisa mais criar uma fila para cada fluxo do Auto Loader, é mais fácil evitar atingir os limites de notificação do provedor de nuvem listados nos recursos de nuvem usados no modo de notificação de arquivo do Auto Loader herdado.
  • O Azure Databricks gerencia automaticamente o ajuste dos requisitos de recursos, portanto, você não precisa ajustar parâmetros como cloudFiles.fetchParallelism.
  • A funcionalidade de limpeza significa que você não precisa se preocupar tanto com o ciclo de vida das notificações criadas na nuvem, como quando um fluxo é excluído ou totalmente atualizado.

Se você usar o Auto Loader no modo de listagem de diretórios, o Databricks recomenda migrar para o modo de notificação de arquivo com eventos de arquivo. Auto Loader com eventos de arquivo oferece melhorias significativas de desempenho. Começa por ativar eventos de ficheiro para a tua localização externa, depois define cloudFiles.useManagedFileEvents a configuração do teu fluxo Auto Loader.

• Modo de notificação de arquivo herdado: você gerencia filas de notificação de arquivos para cada fluxo do Auto Loader separadamente. O Auto Loader configura automaticamente um serviço de notificação e um serviço de fila que monitoriza eventos de arquivos no diretório de entrada.

  Esta é a abordagem legada.

Usar o modo de notificação de arquivo com eventos de arquivo

Esta seção descreve como criar e atualizar fluxos do Auto Loader para usar eventos de arquivo.

Antes de começar

A configuração de eventos de arquivo requer:

• Um espaço de trabalho do Azure Databricks habilitado para o Catálogo Unity.
• Permissão para criar credenciais de armazenamento e objetos de localização externa no Unity Catalog.

Os fluxos do Auto Loader com eventos de arquivo requerem:

• Calcule em Databricks Runtime 14.3 LTS ou superior.

Instruções de configuração

As instruções a seguir se aplicam se você estiver criando novos fluxos do Auto Loader ou migrando fluxos existentes para usar o modo de notificação de arquivo atualizado com eventos de arquivo:

1. Crie uma credencial de armazenamento e um local externo no Catálogo Unity que concedam acesso ao local de origem no armazenamento em nuvem para seus fluxos do Auto Loader.

2. Habilite eventos de arquivo para o local externo. Veja (Recomendado) Ativar eventos de ficheiro para um local externo.

3. Quando você cria um novo fluxo do Auto Loader ou edita um existente para trabalhar com o local externo:

   • Se você tiver fluxos de carregador automático baseados em notificações existentes que consomem dados do local externo, desative-os e exclua os recursos de notificação associados.
   • Certifique-se de que pathRewrites não está definido (esta não é uma opção comum).
   • Revise a lista de configurações que o Auto Loader ignora quando gerencia notificações de arquivo usando eventos de arquivo. Evite-os em novos fluxos do Auto Loader e remova-os dos fluxos existentes que você está migrando para este modo.
   • Defina a opção cloudFiles.useManagedFileEvents para true no seu código Auto Loader.

Por exemplo:

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

Se você estiver usando o Lakeflow Spark Declarative Pipelines e já tiver um pipeline com uma tabela de streaming, atualize-o para incluir a useManagedFileEvents opção:

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'
                   ...
                   );

Configurações do carregador automático não suportadas

As seguintes configurações do Auto Loader não são suportadas quando os fluxos usam eventos de arquivo:

Para as melhores práticas para eventos de ficheiros geridos, consulte Melhores práticas para Auto Loader com eventos de ficheiro.

Limitações do Auto Loader com eventos de arquivo

O serviço de eventos de arquivo otimiza a descoberta de arquivos armazenando em cache os arquivos criados mais recentemente. Se o Auto Loader for executado com pouca frequência, esse cache poderá expirar e o Auto Loader retornará à listagem de diretórios para descobrir arquivos e atualizar o cache. Para evitar esse cenário, invoque o Auto Loader pelo menos uma vez a cada sete dias.

Para obter uma lista geral de limitações em eventos de arquivo, consulte Limitações de eventos de arquivo.

Gerencie filas de notificação de arquivos para cada fluxo do Auto Loader separadamente (legado)

Recursos na nuvem usados no modo de notificação de ficheiros do Auto Loader antigo

O Auto Loader pode configurar notificações de arquivo automaticamente quando definir a opção cloudFiles.useNotifications para true e fornecer as permissões necessárias para criar recursos de nuvem. Além disso, talvez seja necessário fornecer opções adicionais para conceder autorização ao Auto Loader para criar esses recursos.

A tabela a seguir lista os recursos criados pelo Auto Loader para cada provedor de nuvem.

* Auto Loader nomeia os recursos com este prefixo.

** Quantos pipelines de notificação de arquivo simultâneos podem ser iniciados

Se for necessário executar mais fluxos do Auto Loader baseados em notificações de arquivo do que o permitido por esses limites, você poderá usar eventos de arquivo ou um serviço como AWS Lambda, Azure Functions ou Google Cloud Functions para distribuir notificações de uma única fila que escuta um contêiner ou bucket inteiro em filas específicas do diretório.

Eventos de notificação de arquivo legado

Amazon S3 fornece um evento ObjectCreated quando um arquivo é carregado em um bucket do S3, independentemente de ter sido carregado por um carregamento simples ou um carregamento em várias partes.

O Armazenamento Azure Data Lake fornece notificações de eventos diferentes para arquivos que aparecem em seu contêiner de armazenamento.

• O Auto Loader escuta o FlushWithClose evento para processar um arquivo.
• Os fluxos do Auto Loader suportam a RenameFile ação para descobrir arquivos. RenameFile exigem uma solicitação de API ao sistema de armazenamento para obter o tamanho do arquivo renomeado.
• Os fluxos do Auto Loader criados com o Databricks Runtime 9.0 e após suportam a RenameDirectory ação de descoberta de arquivos. RenameDirectory ações exigem solicitações de API ao sistema de armazenamento para listar o conteúdo do diretório renomeado.

O Google Cloud Storage fornece um OBJECT_FINALIZE evento quando um arquivo é carregado, o que inclui substituições e cópias de arquivos. Carregamentos com falha não geram esse evento.

Permissões necessárias para configurar a notificação de arquivo para o Armazenamento do Azure Data Lake e o Armazenamento de Blob do Azure

Você deve ter permissões de leitura para o diretório de entrada. Consulte Armazenamento Blob do Azure.

Para usar o modo de notificação de arquivo, você deve fornecer credenciais de autenticação para configurar e acessar os serviços de notificação de eventos.

Você pode autenticar usando um dos seguintes métodos:

• No Databricks Runtime 16.1 e superior: Credencial de serviço Databricks (recomendado): crie uma credencial de serviço usando uma identidade gerenciada e um conector de acesso Databricks.
• Principal de serviço: crie um aplicativo Microsoft Entra ID (anteriormente Azure Active Directory) e principal de serviço na forma de ID do cliente e segredo do cliente.

Depois de obter as credenciais de autenticação, atribua as permissões necessárias ao conector de acesso do Databricks (para credenciais de serviço) ou à aplicação Microsoft Entra ID (para um principal de serviço).

• Usando funções internas do Azure

  Atribua ao conector de acesso as seguintes funções à conta de armazenamento na qual reside o caminho de entrada:

  • Colaborador: essa função é para configurar recursos em sua conta de armazenamento, como filas e assinaturas de eventos.
  • Colaborador de Dados da Fila de Armazenamento: esta função destina-se a executar operações de fila, como recuperar e excluir mensagens nas filas. Esta função é necessária apenas quando se fornece um principal de serviço sem uma cadeia de conexão.

  Atribua a este conector de acesso a seguinte função ao grupo de recursos relacionado:

  • EventGrid EventSubscription Contributor: esta função destina-se a executar operações de subscrição da Grelha de Eventos do Azure (Grelha de Eventos), tais como criar ou listar subscrições de eventos.

  Para obter mais informações, consulte Utilizar o portal do Azure para atribuir funções do Azure.

• Usando uma função personalizada

  Se você estiver preocupado com as permissões excessivas necessárias para as funções anteriores, poderá criar uma Função Personalizada com pelo menos as seguintes permissões, listadas abaixo no formato JSON da função do 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": []
    }
  ]
  

  Em seguida, você pode atribuir essa função personalizada ao seu conector de acesso.

  Para obter mais informações, consulte Utilizar o portal do Azure para atribuir funções do Azure.

Permissões necessárias para configurar a notificação de arquivo para o Amazon S3

Você deve ter permissões de leitura para o diretório de entrada. Consulte Detalhes da conexão do S3 para obter mais detalhes.

Para usar o modo de notificação de arquivo, anexe o seguinte documento de política JSON ao seu usuário ou função do IAM. Esta função do IAM é necessária para criar uma credencial de serviço com a qual o Auto Loader se autentica.

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

onde:

• <bucket-name>: O nome do bucket do S3 onde seu stream lerá arquivos, por exemplo, auto-logs. Você pode usar * como caracter universal; por exemplo, databricks-*-logs. Para descobrir o bucket subjacente do S3 para o caminho do DBFS, pode listar todos os pontos de montagem do DBFS num notebook executando %fs mounts.
• <region>: A região da AWS onde o bucket do S3 reside, por exemplo, us-west-2. Se não quiser especificar a região, use *.
• <account-number>: o número da conta da AWS que possui o bucket do S3, por exemplo, 123456789012. Se não quiser especificar o número da conta, use *.

A string databricks-auto-ingest-* na especificação SQS e SNS ARN é o prefixo de nome que a origem cloudFiles usa ao criar serviços SQS e SNS. Como o Azure Databricks configura os serviços de notificação na execução inicial do fluxo, você pode usar uma política com permissões reduzidas após a execução inicial (por exemplo, parar o fluxo e reiniciá-lo).

• s3:ListBucket
• s3:GetObject

Permissões reduzidas após a configuração inicial

As permissões de configuração de recursos descritas acima são necessárias apenas durante a execução inicial do fluxo. ** Após a primeira execução, pode mudar para a política IAM seguinte com permissões reduzidas.

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

Permissões necessárias para configurar a notificação de arquivo para GCS

Você deve ter list e get permissões no seu bucket GCS e em todos os objetos. Para obter detalhes, consulte a documentação do Google sobre permissões do IAM.

Para usar o modo de notificação de arquivo, precisa de adicionar permissões para a conta de serviço do GCS e a conta de serviço usada para aceder aos recursos do Google Cloud Pub/Sub.

Adicione a Pub/Sub Publisher função à conta de serviço GCS. Isso permite que a conta publique mensagens de notificação de eventos de seus buckets GCS no Google Cloud Pub/Sub.

Quanto à conta de serviço usada para os recursos do Google Cloud Pub/Sub, você precisa adicionar as seguintes permissões. Essa conta de serviço é criada automaticamente quando você cria uma credencial de serviço Databricks . O suporte a credenciais de serviço está disponível no Databricks Runtime 16.1 e superior.

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

Para fazer isso, você pode criar uma função personalizada do IAM com essas permissões ou atribuir funções GCP pré-existentes para cobrir essas permissões.

Localizando a conta de serviço GCS

No Google Cloud Console do projeto correspondente, navegue até Cloud Storage > Settings. A seção "Conta de serviço de armazenamento em nuvem" contém o e-mail da conta de serviço GCS.

Criação de uma função personalizada do Google Cloud IAM para o modo de notificação de arquivo

No console do Google Cloud para o projeto correspondente, navegue até IAM & Admin > Roles. Em seguida, crie uma função na parte superior ou atualize uma função existente. Na tela de criação ou edição de funções, clique em Add Permissions. Aparece um menu no qual você pode adicionar as permissões desejadas à função.

Configurar ou gerenciar manualmente recursos de notificação de arquivos

Os usuários privilegiados podem configurar ou gerenciar manualmente os recursos de notificação de arquivos.

• Configure os serviços de notificação de arquivos manualmente por meio do provedor de nuvem e especifique manualmente o identificador de fila. Consulte Opções de notificação de arquivo para obter mais detalhes.
• Use APIs do Scala para criar ou gerenciar as notificações e os serviços de enfileiramento, conforme mostrado no exemplo a seguir:

Python

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

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

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

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

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

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

linguagem de programação Scala

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

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

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

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

Use setUpNotificationServices(<resource-suffix>) para criar uma fila e uma subscrição com o nome <prefix>-<resource-suffix> (o prefixo depende do sistema de armazenamento resumido em Recursos na nuvem usados no modo de notificação de arquivos do Auto Loader legado. Se houver um recurso existente com o mesmo nome, o Azure Databricks reutilizará o recurso existente em vez de criar um novo. Essa função retorna um identificador de fila que você pode passar para a cloudFiles origem usando o identificador nas opções de notificação de arquivo. Isso permite que o cloudFiles usuário de origem tenha menos permissões do que o usuário que cria os recursos.

Disponibilize a opção "path" para newManager apenas se estiver a chamar setUpNotificationServices; não é necessária para listNotificationServices ou tearDownNotificationServices. Isso é o mesmo path que você usa ao executar uma consulta de streaming.

A matriz a seguir indica quais métodos de API são suportados em quais Databricks Runtime para cada tipo de armazenamento:

Limpar recursos de notificação de eventos criados pelo Auto Loader

O Auto Loader não destrói automaticamente os recursos de notificação de ficheiros. Para derrubar recursos de notificação de arquivo, você deve usar o gerenciador de recursos de nuvem, conforme mostrado na seção anterior. Você também pode excluir esses recursos manualmente usando a interface do usuário ou as APIs do provedor de nuvem.

Solucionar erros comuns

Esta seção descreve erros comuns ao usar o Auto Loader com o modo de notificação de arquivo e como resolvê-los.

Falha ao criar a assinatura da Grade de Eventos

Se vir a seguinte mensagem de erro quando executa o Carregador Automático pela primeira vez, a Grelha de Eventos não está registada como um Fornecedor de Recursos na subscrição do Azure.

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

Para registrar a Grade de Eventos como um provedor de recursos, faça o seguinte:

1. No portal do Azure, aceda à sua subscrição.
2. Clique em Provedores de Recursos na seção Configurações .
3. Registe o fornecedor Microsoft.EventGrid.

Autorização necessária para executar operações de assinatura da Grade de Eventos

Se vir a seguinte mensagem de erro quando executar o Carregador Automático pela primeira vez, confirme se a função de Colaborador está atribuída à entidade de serviço para o Event Grid e a conta de armazenamento.

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

Cliente de grade de eventos ignora proxy

No Databricks Runtime 15.2 e superior, as conexões de Event Grid no Auto Loader usam configurações de proxy das propriedades do sistema por padrão. No Databricks Runtime 13.3 LTS, 14.3 LTS e 15.0 a 15.2, você pode configurar manualmente as conexões da Grade de Eventos para usar um proxy definindo a propriedade Spark Configspark.databricks.cloudFiles.eventGridClient.useSystemProperties true. Consulte Definir propriedades de configuração do Spark no Azure Databricks.