Skip to content
SparquetSparquet

Your first pipeline

Five minutes, one file, a real dataset on disk. No Studio, no cluster, no database.

  1. Create some data

    Section titled “Create some data”
    Terminal window
    mkdir -p sparquet-demo/data && cd sparquet-demo
    data/orders.csv
    id,customer,country,status,quantity,unit_price,ordered_at
    1,Ada,BR,CONFIRMED,3,25.50,2026-01-04
    2,Linus,PT,CANCELLED,1,80.00,2026-01-05
    3,Grace,BR,CONFIRMED,2,15.00,2026-01-05
    4,Alan,ES,CONFIRMED,7,9.90,2026-01-06
    4,Alan,ES,CONFIRMED,7,9.90,2026-01-06

    The last row is a duplicate on purpose — the pipeline will remove it.

  2. Write the pipeline

    Section titled “Write the pipeline”
    orders.json
    {
      "name": "orders_curated",
      "description": "Clean the order feed and land it as Parquet.",
    

      "input": {
        "format": "csv",
        "path": "data/orders.csv"
      },
    

      "transformations": [
        { "type": "filter", "condition": "status = 'CONFIRMED'" },
        { "type": "cast", "columns": { "quantity": "int", "unit_price": "double", "ordered_at": "date" } },
        { "type": "with_column", "column": "revenue", "expression": "quantity * unit_price" },
        { "type": "drop_duplicates", "columns": ["id"] }
      ],
    

      "validations": {
        "on_failure": "warn",
        "rules": [
          { "type": "not_null", "columns": ["id", "customer"] },
          { "type": "unique", "columns": ["id"] },
          { "type": "row_count", "min": 1 }
        ]
      },
    

      "output": {
        "format": "parquet",
        "path": "out/orders",
        "mode": "overwrite",
        "partition_by": ["country"]
      }
    }

  3. Run it

    Section titled “Run it”
    run.py
    from sparquet import Sparquet
    

    fw = Sparquet(spark={"app_name": "demo", "master": "local[*]"})
    result = fw.run("orders.json")
    

    print(result.summary())
    for check in result.validation_results:
        print(f"  {check.rule_type}: {'ok' if check.passed else check.message}")
    

    fw.stop()
    Terminal window
    python run.py

    Or straight from the terminal, with no Python file at all:

    Terminal window
    python -m sparquet.cli orders.json

  4. Read the result

    Section titled “Read the result”
    • Directorysparquet-demo
      • Directorydata
        • orders.csv
      • Directoryout
        • Directoryorders
          • Directorycountry=BR
            • part-0000….parquet
          • Directorycountry=ES
            • part-0000….parquet
      • orders.json
      • run.py

    Three rows survive: the cancelled order is filtered out and the duplicate is dropped. revenue was computed, and the output is partitioned by country.

What just happened

Section titled “What just happened”

The framework executed the document top to bottom:

Step What ran Where it is defined
1 Read data/orders.csv with header and inferred types input
2 Kept confirmed orders, fixed types, computed revenue, dropped duplicates — in that order transformations
3 Measured three quality rules without changing the data validations
4 Wrote Parquet partitioned by country output

Two details worth internalizing now:

  • Transformations run in the order you write them. Filtering after a select that dropped the column fails — the array is a sequence, not a set.
  • Validations report, they do not clean. not_null counts nulls; it never removes rows. Use a filter for that. See Validations.

Change one thing

Section titled “Change one thing”

Try each of these on the file you just wrote:

Write to two places at once — replace output with outputs:

"outputs": [
  { "format": "parquet", "path": "out/orders", "mode": "overwrite", "partition_by": ["country"] },
  { "format": "csv", "path": "out/orders_report", "mode": "overwrite", "columns": ["id", "customer", "revenue"] }
]

Make the country a parameter — use a placeholder and pass it at run time:

{ "type": "filter", "condition": "country = '{country}'" }
fw.run("orders.json", params={"country": "BR"})

Fail the run on bad data — change on_failure to "fail" and watch the pipeline stop before writing anything.

Next

Section titled “Next”