Arquitetura de fan-in e fan-out nos pipelines do Lakeflow

O fan-in e o fan-out são padrões comuns para criar pipelines de dados escalonáveis e confiáveis. Esta página explica os dois e mostra como implementá-los em pipelines do Lakeflow.

O que são fan-in e fan-out?

Fan-in é um padrão de arquitetura em que os dados de várias fontes são ingeridos e processados em um único pipeline.

As fontes podem incluir:

• Fluxos de eventos em tempo real (por exemplo, Kafka e Kinesis)
• Armazenamento em nuvem (por exemplo, S3, ADLS e Google Cloud Storage)
• Bancos de dados relacionais (por exemplo, PostgreSQL, MySQL e Snowflake)
• Dispositivos IoT (por exemplo, sensores, logs e APIs)

Ao consolidar fluxos de dados diversos em uma única camada de processamento, o fan-in possibilita uma transformação consistente, eliminação de duplicações e enriquecimento de dados antes que eles sejam transmitidos para as etapas seguintes.

Fan-out adota uma abordagem um-para-muitos, roteando um único fluxo de dados processado para vários destinos.

Os destinos podem incluir:

• Tabelas delta para armazenamento estruturado
• Sistemas de alerta em tempo real para detecção de anomalias
• Modelos de machine learning para análise preditiva
• Data warehouses para relatórios e análises
• Filas de mensagens para comunicação assíncrona e processamento desacoplado

Esse padrão garante que cada sistema downstream receba dados no formato necessário, permitindo que as organizações integrem dados de streaming em vários aplicativos de negócios.

Na prática, os pipelines geralmente combinam ambos os padrões. Por exemplo:

• Uma empresa coleta dados de atividade do usuário de vários aplicativos, sites e dispositivos móveis (fan-in).
• Os dados processados são armazenados no Delta Lake para análise histórica, enquanto os alertas em tempo real disparam para atividade incomum (fan-out).

Implementar o fan-in com fluxos de acréscimo

Os pipelines de fan-in mesclam vários fluxos de dados em um destino unificado. Tradicionalmente, isso requer consultas SQL de união complexas e verificação manual de pontos. Os fluxos de acréscimo simplificam isso permitindo que vários fluxos de dados se alimentem diretamente em uma única tabela de streaming sem uniões explícitas ou lógica complexa. Cada fonte é gerenciada de forma independente, permitindo a ingestão e atualizações de dados incrementais.

Por exemplo, use fluxos de acréscimo para consolidar vários tópicos Kafka ou fluxos de dados regionais em uma tabela de destino unificada.

Python

from pyspark import pipelines as dp

dp.create_streaming_table("all_topics")

# Kafka stream from topic1
@dp.append_flow(target="all_topics")
def topic1():
    return spark.readStream.format("kafka") \
        .option("kafka.bootstrap.servers", "host1:port1,...") \
        .option("subscribe", "topic1") \
        .load()

# Kafka stream from topic2
@dp.append_flow(target="all_topics")
def topic2():
    return spark.readStream.format("kafka") \
        .option("kafka.bootstrap.servers", "host1:port1,...") \
        .option("subscribe", "topic2") \
        .load()

SQL

CREATE OR REFRESH STREAMING TABLE all_topics;

CREATE FLOW
  topic1
AS INSERT INTO
  all_topics BY NAME
SELECT * FROM
  read_kafka(bootstrapServers => 'host1:port1,...', subscribe => 'topic1');

CREATE FLOW
  topic2
AS INSERT INTO
  all_topics BY NAME
SELECT * FROM
  read_kafka(bootstrapServers => 'host1:port1,...', subscribe => 'topic2');

Implementar fan-out

Os pipelines de fan-out distribuem dados de uma fonte para várias saídas. Os pipelines do Lakeflow dão suporte a três abordagens, dependendo do caso de uso.

Usar para loops para lógica generalizada

Se sua lógica ETL for idêntica em vários destinos, use o Python para loops para gerar dinamicamente várias tabelas por meio de loops parametrizados. Isso evita a codificação repetitiva e simplifica o dimensionamento de pipeline por meio da configuração.

regions = ["US", "EU", "APAC"]

for region in regions:
    @dp.materialized_view(name=f"orders_{region.lower()}_filtered")
    def filtered_orders(region_filter=region):
        return spark.read.table("combined_orders").filter(f"region = '{region_filter}'")

Usar fluxos independentes para lógica específica do alvo

Quando as transformações de ETL variam significativamente por destino, implemente fluxos de dados independentes. Essa abordagem tem controle preciso e desempenho otimizado adaptado a cada caso de uso.

from pyspark import pipelines as dp
from pyspark.sql.functions import col

# Grouped output
@dp.materialized_view(name="orders_sink")
def region_orders():
    df = spark.read.table("combined_orders").groupBy("region").count()
    # Add additional logic here
    return df

# BI materialized view
@dp.materialized_view(name="orders_bi_materialized")
def orders_bi():
    return spark.read.table("combined_orders").select("order_id", "amount", "region")

# ML feature table
@dp.materialized_view(name="orders_ml_features")
def orders_ml():
    return (
        spark.read.table("combined_orders")
        .withColumn("high_value_order", col("amount") > 1000)
        .select("order_id", "high_value_order", "region")
    )

Usar ForEachBatch para roteamento personalizado

O foreach_batch_sink aplica lógica personalizada a cada microlote, permitindo transformações complexas, mesclagem ou roteamento para vários destinos, incluindo aqueles sem suporte nativo a streaming, como destinos JDBC.

from pyspark import pipelines as dp

@dp.foreach_batch_sink(name="user_events_feb")
def user_events_handler(batch_df, batch_id):
    # Write to Delta table
    batch_df.write.format("delta").mode("append").saveAsTable("my_catalog.my_schema.my_delta_table")

    # Write to JSON files
    batch_df.write.format("json").mode("append").save("/Volumes/path/to/json_target")

@dp.append_flow(target="user_events_feb", name="user_events_flow")
def read_user_events():
    return (
        spark.readStream.format("cloudFiles")
        .option("cloudFiles.format", "json")
        .load("/data/incoming/events")
    )

Padrões comuns de ForEachBatch

O foreach_batch_sink suporta vários padrões. Alguns padrões comuns incluem:

• Fluxo único para o coletor de vários destinos: um único append_flow lê dados de uma fonte de streaming e roteia-os para um foreach_batch_sink. O coletor manipula a gravação em vários destinos (por exemplo, Delta, JSON e sistemas externos). Isso é ideal para casos simples de uso de várias saídas com lógica de transformação compartilhada.

• Vários fluxos para um destino unificado: várias append_flow fontes (por exemplo, diretórios diferentes, formatos, tópicos do Kafka ou APIs externas) convergem para um único foreach_batch_sink. Isso centraliza a lógica de transformação comum, o gerenciamento de saída e o tratamento de erros. Como apenas um ponto de verificação precisa ser mantido, essa abordagem reduz significativamente a complexidade da coordenação. É particularmente útil ao lidar com filas de mensagens, como Kafka ou APIs externas.

• Um fluxo para um coletor (muitos pares independentes): cada append_flow possui um foreach_batch_sink dedicado, estabelecendo relações claras e isoladas entre fontes individuais e seus alvos. Isso é ideal para pipelines com muitos fluxos independentes que exigem lógica de processamento exclusiva, solução de problemas simplificada e tratamento isolado de erros.

Na prática, essas abordagens geralmente se complementam. Por exemplo, use loops para gerar vários fluxos de adição dinamicamente para cenários de fan-in em grande escala e, depois, distribua os resultados usando loops ou foreach_batch_sink para fan-out.

Práticas recomendadas

• Os fluxos de acréscimo exigem que os esquemas de origem estejam alinhados com a tabela de streaming de destino para evitar erros de processamento. Use as expectativas de esquema de pipelines do Lakeflow para detectar e lidar com exceções proativamente, garantindo a consistência do esquema em todo o pipeline.
• Mantenha a lógica de loop bem definida e simples.
• Nomeie cada fluxo e tabela claramente para manter a legibilidade.
• Monitore a utilização de recursos para dimensionar com eficiência e evitar gargalos de desempenho.
• Ao gravar em filas de mensagens, utilize um foreach_batch_sink com um único append_flow que consolide todos os fluxos de entrada. Isso simplifica o estado downstream e o gerenciamento de ponto de verificação.

Limitações

• A interface de usuário de linhagem de pipelines do Lakeflow pode não mostrar métricas e metadados no nível do fluxo para fontes de fluxo de acréscimo novas.
• Expanda a lista de valores usada em um loop 'for', em vez de reduzi-la. Se um conjunto de dados definido anteriormente for omitido em execuções de pipeline subsequentes, ele será removido automaticamente do esquema de destino, o que causa perda de dados não intencional.