Data engineering

Recipes for pipeline fixtures, warehouse loading, and reproducible test data. Each one is a working snippet — copy, run, get the answer on disk. For mental-model context, see How plotsim works.

Pin a deterministic fixture for CI

Check the YAML into tests/fixtures/. The fixture is the config, not the generated tables — same seed in, byte-identical output out.

PythonYAML

from plotsim import create_from_yaml, generate_tables, write_tables

config = create_from_yaml("tests/fixtures/saas_clean.yaml")
tables = generate_tables(config)
write_tables(tables, config, output_dir="tests/_tmp", quiet=True)

plotsim run tests/fixtures/saas_clean.yaml -o tests/_tmp --quiet

On disk: every dim/fact/event/bridge CSV plus config.yaml, validation_report.txt, manifest.json.

Learn more: How plotsim works → Determinism.

Generate Parquet for a data-lake fixture

Switch the format and the rest stays the same. Parquet preserves nested cells and is partition-aware.

PythonYAML

from plotsim import create, generate_tables, write_tables

config = create(
    about="Orders warehouse fixture",
    unit="customer",
    window=("2024-01", "2024-12", "monthly"),
    output={"format": "parquet", "directory": "./lake"},
    metrics=[
        {"name": "engagement",   "type": "score", "polarity": "positive"},
        {"name": "order_volume", "type": "count", "polarity": "positive"},
    ],
    segments=[
        {"name": "loyal",   "count": 60, "archetype": "growth"},
        {"name": "lapsing", "count": 40, "archetype": "decline"},
    ],
)
write_tables(generate_tables(config), config)

about: Orders warehouse fixture
unit: customer
window: { start: "2024-01", end: "2024-12", every: monthly }
output:
  format: parquet
  directory: ./lake

metrics:
  - { name: engagement,   type: score, polarity: positive }
  - { name: order_volume, type: count, polarity: positive }

segments:
  - { name: loyal,   count: 60, archetype: growth }
  - { name: lapsing, count: 40, archetype: decline }

On disk: one .parquet per table. Requires pip install "plotsim[parquet]".

Learn more: Output and scaling → Parquet.

Hive-partition Parquet by year

For multi-year warehouses, partition by a column that exists on at least one table. The engine writes a directory tree <table>/<col>=<value>/... for tables that have the column; others fall back to single files.

PythonYAML

output={
    "format":       "parquet",
    "directory":    "./lake",
    "partition_by": "year",
}

output:
  format:       parquet
  directory:    ./lake
  partition_by: year

On disk: dim_date/year=2023/..., dim_date/year=2024/.... Tables without a year column write as single files.

Learn more: Output and scaling → Partitioned Parquet.

Streaming-ingestion fixture (JSONL)

JSONL is line-delimited JSON — one row per line, nested cells preserved natively, ideal for Kafka/Kinesis/SQS replay.

PythonYAML

output={"format": "jsonl", "directory": "./stream"}

output:
  format: jsonl
  directory: ./stream

On disk: *.jsonl per table. Cells use ISO-8601 dates, JSON null for missing, native nested JSON for struct/array columns.

Learn more: Output and scaling → JSONL.

Database-ready SQL dump

Generate dialect-aware DDL + INSERTs (100 rows per statement) for a target database.

PythonYAML

output={"format": "sql", "directory": "./load", "sql_dialect": "postgresql"}

output:
  format: sql
  directory: ./load
  sql_dialect: postgresql

On disk: single data.sql with CREATE TABLE for every dim/fact/event/bridge, dialect-mapped column types, and batched INSERTs. Primary and foreign keys land as constraints when the actual data is unique.

Learn more: Output and scaling → SQL.

CDC audit on the fact you reconcile

Set cdc: true on the fact whose rows your pipeline reconciles after the fact closes. Quality issues on that fact then flip _op to "U" and bump _updated_at.

PythonYAML

config = create(
    # ... metrics, segments, etc. ...
    facts=[
        {
            "name": "fct_revenue",
            "metrics": ["mrr"],
            "cdc": True,
            "columns": [
                {"name": "date_key",   "type": "ref.dim_date"},
                {"name": "company_id", "type": "ref.dim_company"},
                {"name": "mrr",        "type": "metric.mrr"},
            ],
        },
    ],
    quality=[
        {"table": "fct_revenue", "issue": "late_arrival", "rate": 0.02},
    ],
)

facts:
  - name: fct_revenue
    metrics: [mrr]
    cdc: true
    columns:
      - { name: date_key,   type: ref.dim_date }
      - { name: company_id, type: ref.dim_company }
      - { name: mrr,        type: metric.mrr }

quality:
  - { table: fct_revenue, issue: late_arrival, rate: 0.02 }

On disk: fct_revenue.csv carries _inserted_at, _updated_at, _op columns. Late-arrival rows leave _op="I" (row-level issue); null injection / type mismatch / schema drift on this fact flip _op="U".

Learn more: Designing tables → CDC audit columns.

Denormalized wide tables for warehouse consumers

Some downstream consumers want one wide table per fact with every FK'd dim left-joined onto it. Set denormalized: true:

PythonYAML

output={"format": "csv", "denormalized": True}

output:
  format: csv
  denormalized: true

On disk: <fct>.csv (normalized) and <fct>_wide.csv (denormalized) per fact. Dim columns are prefixed <dim>__; SCD2 dims filter to is_current=True.

Learn more: Output and scaling → Denormalized wide tables.

Inject quality issues for pipeline-failure tests

Six issue types let you teach a pipeline how to fail and recover. The manifest's quality_injections section records every corruption with the original row indices and clean values — your answer key.

PythonYAML

quality=[
    # Null engagement during onboarding blackout
    {"table": "fct_engagement", "issue": "null_injection", "rate": 0.03, "column": "engagement"},
    # Client retries duplicating login events
    {"table": "evt_login", "issue": "duplicate_rows", "rate": 0.015},
    # Late-arriving revenue reconciliation
    {"table": "fct_revenue", "issue": "late_arrival", "rate": 0.02},
    # Schema drift on order_total (v2 column)
    {"table": "fct_orders", "issue": "schema_drift", "rate": 0.04, "column": "order_total"},
    # Black Friday spike at period 11
    {
        "table":  "evt_login",
        "issue":  "volume_anomaly",
        "rate":   0.5,
        "mode":   "spike",
        "period": 11,
    },
]

quality:
  - { table: fct_engagement, issue: null_injection, rate: 0.03, column: engagement }
  - { table: evt_login,      issue: duplicate_rows, rate: 0.015 }
  - { table: fct_revenue,    issue: late_arrival,   rate: 0.02 }
  - { table: fct_orders,     issue: schema_drift,   rate: 0.04, column: order_total }
  - table:  evt_login
    issue:  volume_anomaly
    rate:   0.5
    mode:   spike
    period: 11

On disk: corrupted rows in the named tables; manifest.json carries a quality_injections list with the original indices and clean values per (issue_index, table, column).

Learn more: Adding realism → Quality issues.

Multi-source overlap for entity-resolution tests

When two upstream systems both record "customer" but disagree on identifiers, names, and attributes, plotsim emits drifted per-source dim copies alongside the canonical dim.

PythonYAML

sources=[
    {"name": "crm",     "key_scheme": "uuid_short", "name_drift_rate": 0.10, "attribute_drift_rate": 0.05},
    {"name": "billing", "key_scheme": "numeric",    "name_drift_rate": 0.05, "attribute_drift_rate": 0.15},
]

sources:
  - { name: crm,     key_scheme: uuid_short, name_drift_rate: 0.10, attribute_drift_rate: 0.05 }
  - { name: billing, key_scheme: numeric,    name_drift_rate: 0.05, attribute_drift_rate: 0.15 }

On disk: dim_<entity>_crm.csv and dim_<entity>_billing.csv alongside the canonical dim_<entity>.csv. Each per-source row has a drifted name, attributes, and a key in the per-source scheme. manifest.source_entity_mappings records the canonical ↔ per-source mapping per (entity, source).

Learn more: Running experiments → Multi-source overlap.

Raise the cell-budget cap for a large run

The validator raises if total cells exceed 2,000,000 without opt-in:

plotsim run config.yaml -o ./output --allow-large-dataset

Or in the config:

PythonYAML

output={"format": "parquet", "cell_budget": 5_000_000}

output:
  format:      parquet
  cell_budget: 5000000

The hard ceiling is 50,000,000 cells and is not overridable.

Learn more: Output and scaling → Cell-budget gate.

Log-file companion for event tables

For pipelines that ingest log files, set log_format on an event table — a Python str.format template whose placeholders match column names. The writer emits <event_name>.log alongside the CSV/Parquet.

PythonYAML

events=[
    {
        "name":         "evt_login",
        "trigger":      "proportional",
        "driver":       "engagement",
        "scale":        10.0,
        "log_format":   "{event_ts} [INFO] {company_id} login {event_id}",
        "log_filename": "logins.log",
        "columns":      [
            {"name": "event_id",   "type": "id"},
            {"name": "date_key",   "type": "ref.dim_date"},
            {"name": "company_id", "type": "ref.dim_company"},
            {"name": "event_ts",   "type": "timestamp"},
        ],
    },
]

events:
  - name: evt_login
    trigger: proportional
    driver: engagement
    scale: 10.0
    log_format: "{event_ts} [INFO] {company_id} login {event_id}"
    log_filename: logins.log
    columns:
      - { name: event_id,   type: id }
      - { name: date_key,   type: ref.dim_date }
      - { name: company_id, type: ref.dim_company }
      - { name: event_ts,   type: timestamp }

On disk: evt_login.csv plus logins.log (one line per event).

Learn more: Output and scaling → Log-file companion writer.