Ingerir dados de uma API em pipelines

Ingerir de uma API significa extrair dados por HTTP de um serviço web, normalmente como JSON paginado, em vez de ler de um ficheiro ou de uma base de dados. Ao contrário dos ficheiros ou de um barramento de mensagens, não existe uma API genérica incorporada, por isso trata da autenticação, paginação e limites de taxa tu próprio. As pipelines do Lakeflow admitem três padrões de ingestão a partir de qualquer API. Qual deles encaixa depende do teu volume e das tuas necessidades de atualização.

Pré-requisitos

• Um oleoduto. Para criar uma, consulte os tutoriais de pipelines do Lakeflow.
• Credenciais API, como um token ou chave, armazenadas como um segredo do Azure Databricks. Nunca codifique credenciais no código-fonte do pipeline. Consulte Gestão secreta.
• Acesso à rede a partir do seu pipeline compute até ao endpoint da API.
• Familiaridade com tabelas de streaming e vistas materializadas, os tipos de conjunto de dados que estes padrões produzem. Consulte Tabelas de transmissão em fluxo e Vistas materializadas.

Escolha um padrão

Não existe uma origem genérica nativa para APIs REST em pipelines, por isso, quando obténs dados de uma API arbitrária, escolhe um de três padrões com base no volume de dados e na frequência com que os importas:

Padrão 1: Puxões periódicos como uma vista materializada

Para volumes de dados pequenos a médios, obtidos uma vez em cada execução do pipeline, escreva uma função Python que chame a API e devolva um DataFrame do Spark. Uma vez que o conjunto de dados é uma vista materializada, o fluxo de processamento volta a executar a função na totalidade e de forma idempotente sempre que é atualizado.

Os passos seguintes mostram-lhe como construir uma vista materializada com puxagens periódicas:

1. Guarda o token da API em segredo e depois mapeia-o para uma propriedade de configuração do Spark nas definições do pipeline para que o código do pipeline o possa ler. Adicione a propriedade ao bloco spark_conf da configuração do cluster da canalização:

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

   O código no passo seguinte lê este valor com spark.conf.get("api.token"). Para saber mais sobre a configuração de segredos nas definições do pipeline, veja Acesso Seguro às credenciais de armazenamento com segredos num pipeline.

2. Defina uma vista materializada que chama a API e devolve a resposta como 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 da paginação dentro da função, fazendo um loop sobre páginas e concatenando os resultados antes de devolver 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 espera progressiva no pedido para aumentar a resiliência.

Este padrão relê a resposta completa da API em cada atualização do pipeline, por isso usa-o apenas quando a carga útil estiver limitada. Para leituras incrementais, use o padrão 2.

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

Se precisar de consultar as APIs incrementalmente com seguimento de offsets, implemente uma fonte de dados personalizada com a API Python de Fonte de Dados do Spark. Isto proporciona-lhe a semântica correta de streaming, incluindo progresso com checkpoints e leituras incrementais, para que, ao reiniciar, o processo retome a partir do último offset em vez de voltar a consultar toda a API.

Os passos seguintes mostram-lhe como ingerir a partir de uma fonte de dados personalizada:

1. Implemente um DataSource e DataSourceStreamReader que chamam a API e controlam o offset de leitura. Para detalhes sobre a criação de uma fonte de dados personalizada, consulte PySpark fontes de dados personalizadas.

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

   spark.dataSource.register(MyApiDataSource)
   

3. Leia da fonte registada numa 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 uma tarefa agendada e o Auto Loader

Um padrão comum de produção é separar a chamada API do pipeline. Um trabalho agendado envia as respostas brutas da API como ficheiros num volume do Unity Catalog, e o pipeline capta-as com o Auto Loader. Isto isola particularidades específicas da API, como paginação e limites de taxa, da tua lógica de transformação declarativa, e dá-te o rastreio de ficheiros exato uma vez do Auto Loader gratuitamente.

Os passos seguintes mostram-lhe como desacoplar a ingestão através de uma tarefa agendada:

1. Escreve um caderno ou script que chame a API e escreva as respostas JSON brutas para um volume do Unity Catalog. Leia as credenciais da API armazenadas num segredo. Consulte Gestão secreta.

   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 caderno ou script para correr sozinho com Lakeflow Jobs. Veja Lakeflow Jobs.

3. No seu pipeline, defina uma tabela de streaming que leia os ficheiros 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 fiável de ficheiros com o Auto Loader, consulte Carregar ficheiros a partir do armazenamento de objetos na nuvem e O que é o Auto Loader?.

Boas práticas para a ingestão de APIs

• Mantenha segredos fora do código-fonte. Armazene tokens e chaves API em escopos secretos do Azure Databricks e leia-os em tempo de execução. Consulte Gestão secreta.
• Valide as respostas desde cedo. Adicione expectativas sobre as linhas ingeridas para detetar respostas API malformadas antes que fluam para jusante.
• Trate da paginação e dos limites de taxa. Percorre as páginas e adiciona novas tentativas com atraso progressivo para que uma falha transitória não faça falhar toda a atualização.

Recursos adicionais

• Carregar dados em pipelines
• Conectores gerenciados no Lakeflow Connect
• Fontes de dados personalizadas PySpark
• Gerir a qualidade dos dados através das expectativas do pipeline