Ingerir dados de uma API em pipelines

Ingerir dados de uma API significa obter dados via HTTP de um serviço web, geralmente em JSON paginado, em vez de lê-los de um arquivo ou de um banco de dados. Ao contrário de arquivos ou de um barramento de mensagens, não há uma fonte de API genérica integrada, então você mesmo precisa cuidar da autenticação, da paginação e dos limites de requisições. Os pipelines do Lakeflow oferecem suporte a três padrões para ingestão a partir de uma API arbitrária. Qual deles é o mais adequado depende do seu volume e das suas necessidades de atualização.

Pré-requisitos

• Um oleoduto. Para criar um, consulte Tutoriais de pipelines do Lakeflow.
• Credenciais da API, como token ou chave, armazenadas como um segredo do Azure Databricks. Nunca codifique credenciais diretamente no código-fonte do pipeline. Confira Gerenciamento de segredos.
• Acesso à rede do seu processamento de pipeline até o ponto de extremidade da API.
• Familiaridade com tabelas de streaming e visualizações materializadas, os tipos de conjunto de dados que esses padrões produzem. Veja Tabelas de streaming e visualizações materializadas.

Escolher um padrão

Não há uma fonte genérica nativa de API REST em pipelines; portanto, ao extrair dados de uma API arbitrária, escolha um de três padrões com base no volume de dados e na frequência de ingestão:

Padrão 1: Extrações periódicas como uma visão materializada

Para cargas úteis pequenas a médias extraídas uma vez por execução de pipeline, escreva uma função em Python que chame a API e retorne um DataFrame Spark. Como o conjunto de dados é uma visualização materializada, o pipeline executa novamente a função por completo, de forma idempotente, sempre que o pipeline é atualizado.

Os passos a seguir mostram como construir uma visualização materializada com puxadas periódicas:

1. Armazene o token da API em segredo, depois o mapeie para uma propriedade de configuração do Spark nas configurações do seu pipeline para que o código do pipeline possa lê-lo. Adicione a propriedade ao bloco spark_conf da configuração de cluster do pipeline:

   {
     "clusters": [
       {
         "spark_conf": {
           "api.token": "{{secrets/<scope-name>/<secret-name>}}"
         }
       }
     ]
   }
   

   O código no próximo passo lê esse valor com spark.conf.get("api.token"). Para saber mais sobre como configurar segredos nas configurações do pipeline, veja Acesso Seguro a credenciais de armazenamento com segredos em um pipeline.

2. Defina uma visão materializada que chama a API e retorna a resposta como um DataFrame:

   import requests
   from pyspark import pipelines as dp
   from pyspark.sql import Row
   
   @dp.materialized_view(
       name="exchange_rates_bronze",
       comment="Daily FX rates pulled from a public REST API",
   )
   def exchange_rates_bronze():
       resp = requests.get(
           "https://api.example.com/v1/rates",
           params={"base": "USD"},
           headers={"Authorization": f"Bearer {spark.conf.get('api.token')}"},
           timeout=30,
       )
       resp.raise_for_status()
       rates = resp.json()["rates"]
       rows = [Row(currency=k, rate=float(v), as_of_date=resp.json()["date"]) for k, v in rates.items()]
       return spark.createDataFrame(rows)
   

3. Trate a paginação dentro da função, percorrendo as páginas em um loop e concatenando os resultados antes de retornar o DataFrame:

   import requests
   from pyspark import pipelines as dp
   from pyspark.sql import Row
   
   @dp.materialized_view(
       name="customers_bronze",
       comment="Customers pulled from a paginated REST API",
   )
   def customers_bronze():
       token = spark.conf.get("api.token")
       rows = []
       url = "https://api.example.com/v1/customers"
       while url:  # follow the API's next-page cursor until exhausted
           resp = requests.get(
               url,
               headers={"Authorization": f"Bearer {token}"},
               timeout=30,
           )
           resp.raise_for_status()
           payload = resp.json()
           rows.extend(Row(**record) for record in payload["data"])
           url = payload.get("next")  # None on the last page
       return spark.createDataFrame(rows)
   

   Adicione lógica de repetição e retirada na solicitação para maior resiliência.

Esse padrão lê novamente a resposta completa da API a cada atualização do pipeline, então use-o apenas quando o volume de dados for limitado. Para leituras incrementais, use o padrão 2.

Padrão 2: APIs de alto volume ou streaming com a API de Fonte de Dados Python

Para APIs que exigem consulta incremental com controle de offset, implemente uma fonte de dados personalizada usando a API Python de fontes de dados do Spark. Isso oferece a você a semântica correta de streaming, incluindo progresso registrado em checkpoints e leituras incrementais, para que uma reinicialização retome a partir do último deslocamento em vez de extrair novamente todos os dados da API.

Os passos a seguir mostram como ingerir de uma fonte de dados personalizada:

1. Implemente um DataSource e DataSourceStreamReader que chamem a API e acompanhem o deslocamento de leitura. Para detalhes sobre como criar uma fonte de dados personalizada, veja PySpark fontes de dados personalizadas.

2. Registre a fonte de dados para que o pipeline possa referencia-la pelo nome do formato:

   spark.dataSource.register(MyApiDataSource)
   

3. Leia a partir da fonte registrada em uma tabela de streaming:

   from pyspark import pipelines as dp
   
   @dp.table(name="events_bronze")
   def events_bronze():
       return spark.readStream.format("my_api_source").load()
   

Padrão 3: Desacoplar a ingestão com um trabalho agendado e o Auto Loader

Um padrão comum de produção é separar a chamada de API do pipeline. Um trabalho agendado envia as respostas brutas da API como arquivos em um volume do Unity Catalog, e o pipeline as capta com o Auto Loader. Isso isola da sua lógica declarativa de transformação peculiaridades específicas da API, como paginação e limites de taxa, e oferece, sem custo adicional, o rastreamento de arquivos exatamente uma vez do Carregador Automático.

Os passos a seguir mostram como desvincular a ingestão de um trabalho agendado:

1. Escreva um notebook ou script que chame a API e escreva as respostas JSON brutas em um volume do Unity Catalog. Leia as credenciais da API a partir de um segredo. Confira Gerenciamento de segredos.

   import requests, json, time
   
   token = dbutils.secrets.get(scope="<scope-name>", key="<secret-name>")
   volume_path = "/Volumes/main/raw/landing/api_events"
   
   resp = requests.get(
       "https://api.example.com/v1/events",
       headers={"Authorization": f"Bearer {token}"},
       timeout=30,
   )
   resp.raise_for_status()
   # One file per run; the pipeline's Auto Loader tracks which files it has ingested.
   with open(f"{volume_path}/events_{int(time.time())}.json", "w") as f:
       json.dump(resp.json()["data"], f)
   

2. Programe o notebook ou script para rodar sozinho com Lakeflow Jobs. Consulte Trabalhos do Lakeflow.

3. No seu pipeline, defina uma tabela de streaming que leia os arquivos aterrados com o Auto Loader:

   from pyspark import pipelines as dp
   
   @dp.table(name="api_events_bronze")
   def api_events_bronze():
       return (
           spark.readStream.format("cloudFiles")
               .option("cloudFiles.format", "json")
               .load("/Volumes/main/raw/landing/api_events")
       )
   

Para saber mais sobre a ingestão confiável de arquivos com o Auto Loader, veja Carregar arquivos a partir do armazenamento de objetos na nuvem e O que é o Auto Loader?.

Melhores práticas para a ingestão de APIs

• Mantenha segredos fora do código-fonte. Armazene tokens e chaves de API em escopos secretos do Azure Databricks e leia-os em tempo de execução. Confira Gerenciamento de segredos.
• Valide as respostas desde o início. Adicione expectativas nas linhas ingeridas para detectar respostas de API malformadas antes que elas fluam downstream.
• Gerencie a paginação e os limites de taxa. Percorra as páginas e adicione repetição com retirada para que uma falha transitória não comprometa toda a atualização.

Recursos adicionais

• Carregar dados em pipelines
• Conectores gerenciados no Lakeflow Connect
• Fontes de dados personalizadas do PySpark
• Gerenciar a qualidade dos dados com as expectativas do pipeline