Inicio Rapido

Esta guia te lleva paso a paso a crear y ejecutar un pipeline que ingesta un archivo CSV hacia Parquet en tu sistema de archivos local.

1. Instalar Pipe y el tap de CSV

pip install dataspoc-pipe
pip install tap-csv

2. Crear un CSV de ejemplo

mkdir -p /tmp/sample-data
cat > /tmp/sample-data/orders.csv << 'EOF'
id,customer,product,amount,created_at
1,Alice,Widget,29.99,2025-01-15
2,Bob,Gadget,49.99,2025-01-16
3,Charlie,Widget,29.99,2025-01-17
4,Alice,Gizmo,19.99,2025-01-18
5,Diana,Gadget,49.99,2025-01-19
EOF

3. Inicializar el directorio de configuracion

dataspoc-pipe init

Salida esperada:

Structure created at /home/you/.dataspoc-pipe
Next step: dataspoc-pipe add <name>

Esto crea la estructura de directorios:

~/.dataspoc-pipe/
  config.yaml       # Global defaults
  sources/           # Tap configuration files
  pipelines/         # Pipeline YAML definitions
  transforms/        # Optional Python transforms

4. Crear un pipeline

dataspoc-pipe add my-first-pipeline

El asistente interactivo preguntara:

Creating pipeline: my-first-pipeline

Taps with template: tap-csv, tap-postgres, tap-mysql, ...

Tap Singer: tap-csv
Template found for tap-csv
Source config: /home/you/.dataspoc-pipe/sources/my-first-pipeline.json
Destination bucket (e.g.: s3://my-bucket, file:///tmp/lake): file:///tmp/lake
Base path in bucket [raw]: raw
Compression (zstd, snappy, gzip, none) [zstd]: zstd
Enable incremental extraction? [y/N]: N
Cron expression for scheduling (empty to skip):

Pipeline saved at /home/you/.dataspoc-pipe/pipelines/my-first-pipeline.yaml

Next steps:
  1. Edit source config: /home/you/.dataspoc-pipe/sources/my-first-pipeline.json
  2. Validate: dataspoc-pipe validate my-first-pipeline
  3. Run: dataspoc-pipe run my-first-pipeline

5. Editar la configuracion de la fuente

Abre ~/.dataspoc-pipe/sources/my-first-pipeline.json y apuntalo al CSV:

{
  "csv_files_definition": [
    {
      "entity": "orders",
      "path": "/tmp/sample-data/orders.csv",
      "keys": ["id"]
    }
  ]
}

6. Revisar el YAML del pipeline

El archivo generado en ~/.dataspoc-pipe/pipelines/my-first-pipeline.yaml se ve asi:

source:
  tap: tap-csv
  config: /home/you/.dataspoc-pipe/sources/my-first-pipeline.json
destination:
  bucket: file:///tmp/lake
  path: raw
  compression: zstd
  partition_by: _extraction_date
incremental:
  enabled: false
schedule:
  cron: null

7. Validar

dataspoc-pipe validate my-first-pipeline

Salida esperada:

Validating: my-first-pipeline
  Bucket OK: file:///tmp/lake
  Tap OK: tap-csv found in PATH

8. Ejecutar el pipeline

dataspoc-pipe run my-first-pipeline

Salida esperada:

Running: my-first-pipeline
  orders: 5 records...
Done! 5 records in 1 stream(s)
  orders: 5

9. Verificar el estado

dataspoc-pipe status

                   Pipelines
┌───────────────────┬─────────────────────┬─────────┬──────────┬─────────┐
│ Pipeline          │ Last Run            │ Status  │ Duration │ Records │
├───────────────────┼─────────────────────┼─────────┼──────────┼─────────┤
│ my-first-pipeline │ 2025-01-20T10:30:00 │ success │ 1.2s     │ 5       │
└───────────────────┴─────────────────────┴─────────┴──────────┴─────────┘

10. Ver el log de ejecucion

dataspoc-pipe logs my-first-pipeline

{
  "pipeline": "my-first-pipeline",
  "status": "success",
  "started_at": "2025-01-20T10:30:00Z",
  "finished_at": "2025-01-20T10:30:01Z",
  "duration_seconds": 1.2,
  "total_records": 5,
  "streams": {
    "orders": 5
  }
}

11. Inspeccionar el manifiesto

dataspoc-pipe manifest file:///tmp/lake

El manifiesto muestra todas las tablas que Pipe ha escrito en este bucket, incluyendo esquemas, conteos de registros y marcas de tiempo. Este es el catalogo que las herramientas posteriores como DataSpoc Lens usan para descubrir los datos disponibles.

12. Inspeccionar el bucket

/tmp/lake/
  .dataspoc/
    manifest.json
    state/my-first-pipeline/state.json
    logs/my-first-pipeline/2025-01-20T103000Z.json
  raw/
    csv/orders/
      dt=2025-01-20/
        orders_0000.parquet

El archivo Parquet esta listo para consultar con DuckDB, Pandas, Polars o DataSpoc Lens.

Siguientes pasos

• Configuracion — referencia completa del YAML del pipeline
• Extraccion Incremental — solo obtener datos nuevos
• Transformaciones — limpiar datos durante la ingesta
• Almacenamiento Multi-Cloud — escribir en S3, GCS o Azure
• MCP Server — conectar agentes de IA a Pipe