Migrando do Fivetran para o DataSpoc Pipe: Guia Passo a Passo

Host: db.company.com
Port: 5432
Database: production
Schema: public
Tables: orders, customers, products
Sync mode: Incremental

pipeline: postgres-production
source:
  tap: tap-postgres
  config:
    host: "${POSTGRES_HOST}"
    port: 5432
    database: production
    user: "${POSTGRES_USER}"
    password: "${POSTGRES_PASSWORD}"
    filter_schemas: ["public"]
    filter_tables: ["orders", "customers", "products"]
    replication_method: "LOG_BASED"  # or INCREMENTAL

destination:
  bucket: "s3://my-data-lake"
  path: "raw/postgres"
  format: parquet

pipeline: stripe-data
source:
  tap: tap-stripe
  config:
    client_secret: "${STRIPE_SECRET_KEY}"
    start_date: "2025-01-01T00:00:00Z"
    account_id: "${STRIPE_ACCOUNT_ID}"

destination:
  bucket: "s3://my-data-lake"
  path: "raw/stripe"
  format: parquet

pipeline: google-sheets
source:
  tap: tap-google-sheets
  config:
    oauth_credentials:
      client_id: "${GOOGLE_CLIENT_ID}"
      client_secret: "${GOOGLE_CLIENT_SECRET}"
      refresh_token: "${GOOGLE_REFRESH_TOKEN}"
    spreadsheet_id: "1BxiMVs0XRA5nFMdKvBdBZjgmUUqptlbs74OgVE2upms"
    start_date: "2025-01-01T00:00:00Z"

destination:
  bucket: "s3://my-data-lake"
  path: "raw/google_sheets"
  format: parquet

pipeline: hubspot-crm
source:
  tap: tap-hubspot
  config:
    access_token: "${HUBSPOT_ACCESS_TOKEN}"
    start_date: "2025-01-01T00:00:00Z"

destination:
  bucket: "s3://my-data-lake"
  path: "raw/hubspot"
  format: parquet

from dataspoc_lens import LensClient

lens = LensClient()

# Compare row counts with Fivetran dashboard
for table in ["raw_postgres_orders", "raw_postgres_customers", "raw_postgres_products"]:
    count = lens.query(f"SELECT COUNT(*) as cnt FROM {table}")
    print(f"{table}: {count['cnt'].iloc[0]} rows")

# PostgreSQL — every 6 hours (matches Fivetran schedule)
0 */6 * * * /usr/local/bin/dataspoc-pipe run /opt/pipelines/postgres-production.yaml >> /var/log/pipe/postgres.log 2>&1

# Stripe — every hour
0 * * * * /usr/local/bin/dataspoc-pipe run /opt/pipelines/stripe-data.yaml >> /var/log/pipe/stripe.log 2>&1

# Google Sheets — daily at 2 AM
0 2 * * * /usr/local/bin/dataspoc-pipe run /opt/pipelines/google-sheets.yaml >> /var/log/pipe/sheets.log 2>&1

# HubSpot — every 6 hours
30 */6 * * * /usr/local/bin/dataspoc-pipe run /opt/pipelines/hubspot-crm.yaml >> /var/log/pipe/hubspot.log 2>&1

# Simple runner script with error handling
import subprocess
import sys
from datetime import datetime

pipelines = [
    "postgres-production.yaml",
    "stripe-data.yaml",
    "google-sheets.yaml",
    "hubspot-crm.yaml",
]

results = []
for pipeline in pipelines:
    start = datetime.now()
    result = subprocess.run(
        ["dataspoc-pipe", "run", f"/opt/pipelines/{pipeline}"],
        capture_output=True, text=True
    )
    elapsed = (datetime.now() - start).total_seconds()
    status = "OK" if result.returncode == 0 else "FAILED"
    results.append({"pipeline": pipeline, "status": status, "seconds": elapsed})
    if result.returncode != 0:
        print(f"FAILED: {pipeline}\n{result.stderr}", file=sys.stderr)

# Print summary
for r in results:
    print(f"{r['status']:6s} {r['pipeline']:40s} ({r['seconds']:.1f}s)")

from dataspoc_lens import LensClient

lens = LensClient()

# Compare Fivetran output (in warehouse) vs Pipe output (in lake)
# You can query both if your warehouse data is also accessible

# Check Pipe output
pipe_count = lens.query("SELECT COUNT(*) as cnt FROM raw_postgres_orders")
print(f"Pipe: {pipe_count['cnt'].iloc[0]} rows")

# If they match for 14 days straight, you are safe to cut over

[ ] Inventariar todos os conectores Fivetran
[ ] Encontrar Singer tap para cada fonte
[ ] Criar configuração YAML do Pipe para cada fonte
[ ] Testar cada pipeline com uma execução completa
[ ] Verificar contagens de linhas com o Fivetran
[ ] Configurar agendamento com cron
[ ] Rodar em paralelo por 2 semanas
[ ] Comparar contagens de linhas diariamente
[ ] Fazer a troca para o Pipe
[ ] Monitorar por 48 horas
[ ] Desabilitar conectores Fivetran
[ ] Cancelar assinatura do Fivetran
[ ] Atualizar documentação
[ ] Notificar stakeholders