11-02-01 — Data Factory Pipelines e Dataflows Gen2

Data Factory Pipelines

Os Pipelines do Fabric são similares ao Azure Data Factory v2 — mas nativos ao Fabric, sem custo adicional de serviço. Um pipeline é um grafo de atividades executadas em sequência ou paralelo.

Atividades principais

Configurando um pipeline via código (API)

# Criando pipeline via Fabric REST API (Python)
import requests
import json

headers = {
    "Authorization": f"Bearer {access_token}",
    "Content-Type": "application/json"
}

pipeline_definition = {
    "name": "pipeline_sap_para_bronze",
    "properties": {
        "activities": [
            {
                "name": "CopySAP",
                "type": "Copy",
                "inputs": [{"referenceName": "SAPDataset", "type": "DatasetReference"}],
                "outputs": [{"referenceName": "BronzeDataset", "type": "DatasetReference"}],
                "typeProperties": {
                    "source": {"type": "SapTableSource", "partitionOption": "PartitionOnInt"},
                    "sink": {"type": "LakehouseTableSink", "tableActionOption": "Append"}
                }
            }
        ]
    }
}

response = requests.post(
    f"https://api.fabric.microsoft.com/v1/workspaces/{workspace_id}/items",
    headers=headers,
    json={"type": "Pipeline", "displayName": "pipeline_sap_bronze", "definition": pipeline_definition}
)

Dataflows Gen2

Dataflows Gen2 são transformações visuais baseadas em Power Query — a mesma linguagem M do Excel e Power BI. Analistas de negócio conseguem criar transformações sem escrever Spark ou SQL.

Casos de uso ideais para Dataflows Gen2

• Analistas que sabem Power Query mas não Python/Spark
• Transformações simples: filtrar, renomear colunas, mudar tipos, fazer joins básicos
• Ingestão de Excel e SharePoint Lists
• Prototipação rápida antes de codificar em PySpark

Limitações dos Dataflows

• Performance inferior ao Spark para volumes > 100 GB
• Transformações complexas (ML, NLP) não são possíveis
• Debugging mais difícil que notebooks Python
• Não suporta Delta Lake time travel nativamente

Scheduling e monitoramento

# Monitorando execuções de pipeline via API
response = requests.get(
    f"https://api.fabric.microsoft.com/v1/workspaces/{workspace_id}/items/{pipeline_id}/jobs",
    headers=headers
)

jobs = response.json()["value"]
for job in jobs:
    print(f"Job {job['id']}: {job['status']} - Duração: {job['duration']}ms")
    if job['status'] == 'Failed':
        print(f"  Erro: {job['failureReason']['message']}")