databricks-lakeflow-connect

Files (1)

• SKILL.md

SKILL.md (first lines)

# LakeFlow Connect

LakeFlow Connect is Databricks' **managed ingestion** product for streaming data from external sources into Unity Catalog using Delta Live Tables (DLT) pipelines. It is **GA** as of 2025, replacing the need for third-party ETL tools (Fivetran, Airbyte) for supported source types.

> LakeFlow Connect é diferente de LakeFlow Pipelines (DLT). Connect = ingestão de fontes externas. Pipelines = transformação Spark/SQL dentro do Databricks.

---

## When to Use

| Use Case | LakeFlow Connect |
|----------|-----------------|
| CDC de banco relacional (PostgreSQL, MySQL, SQL Server, Oracle) | ✓ Ideal |
| Ingestão de SaaS (Salesforce, ServiceNow, Workday, NetSuite) | ✓ Ideal |
| Streaming de eventos Kafka / Kinesis | ✗ Usar Auto Loader ou structured streaming diretamente |
| Arquivos em cloud storage (S3, ADLS, GCS) | ✗ Usar Auto Loader |
| APIs REST customizadas | ✗ Construir pipeline Python custom |

---

## Architecture Overview

```
Source (DB / SaaS)
       │
       ▼
LakeFlow Connect (managed connector)
       │  ← DLT pipeline gerenciado pela Databricks
       ▼
Bronze table (Unity Catalog) — raw / append-only
       │
       ▼
Silver table — via DLT transformation (opcional, mesmo pipeline)
```

- Connectors rodam como **DLT pipelines** no workspace Databricks
- Dados são escritos em **Delta tables no Unity Catalog**
- Suporte a **full refresh** e **incremental** (CDC via log-based replication)
- Monitoramento via **Pipeline Events** e métricas nativas do DLT

---

## Supported Sources (GA — Abril 2026)

### Database Sources (CDC via log-based replication)

| Source | CDC Method | Notes |
|--------|-----------|-------|
| PostgreSQL | Logical replication (pgoutput) | Requer `wal_level = logical` |
| MySQL | Binary log (binlog) | Requer `binlog_format = ROW` |
| SQL Server | CDC / change tracking | Requer SQL Server CDC habilitado |
| Oracle | LogMiner | Requer supplemental logging |

### SaaS Sources

| Source | Ingestion Type | Notes |
|--------|---------------|-------|
| Salesforce | Bulk API 2.0 + incremental | Objects e Custom Objects |
| ServiceNow | Table API incremental | Suporte a campos sys_updated_on |
| Workday | Raas Reports incremental | Requer Workday ISU |
| NetSuite | SuiteQL incremental | Requer token-based auth |
| Google Analytics 4 | Batch daily | 28 dias de lookback |
| HubSpot | REST API incremental | Deals, contacts, companies |

---

## Quick Start — Database CDC (PostgreSQL)

### 1. Pré-requisitos no PostgreSQL

```sql
-- No PostgreSQL, habilitar logical replication
ALTER SYSTEM SET wal_level = logical;
ALTER SYSTEM SET max_replication_slots = 10;
SELECT pg_reload_conf();

-- Criar usuário de replicação
CREATE USER lakeflow_user WITH REPLICATION LOGIN PASSWORD 'secure-password';
GRANT SELECT ON ALL TABLES IN SCHEMA public TO lakeflow_user;

-- Habilitar replicação para tabelas específicas
CREATE PUBLICATION lakeflow_pub FOR TABLE orders, customers, products;
```

### 2. Criar Connection no Databricks

```python
# Via Databricks SDK
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

# Criar connection para PostgreSQL
connection = w.connections.create(
    name="prod-postgresql",
    connection_type="POSTGRESQL",
    options={
        "host": "prod-db.example.com",
        "port": "5432",
        "database": "mydb",
        "user": "lakeflow_user",
        "password": "{{secrets/lakeflow/pg-password}}",  # usar Databricks Secrets
    }
)
print(f"Connection ID: {connection.name}")
```

### 3. Criar Ingestion Pipeline

```python
# Criar pipeline de ingestão LakeFlow Connect
pipeline = w.pipelines.create(
    name="postgresql-cdc-pipeline",
    ingestion_definition={
        "connection_name": "prod-postgresql",
        "objects": [
            {
                "schema": {
                    "source_catalog": None,  # não aplicável para DB sources

From ThomazRossito/data-agents, MIT.

More from ThomazRossito/data-agents