pipelines

name	pipelines
description	Use for data, ML, CI, and monitoring pipelines where the task is to assemble inputs, transform records, preserve dependencies, process streams, optimize workflow execution, or produce durable outputs and run statistics.

Pipelines

When To Use

Use this skill when the core task is a sequence of dependent operations: ingestion, transformation, enrichment, joining, windowing, deduplication, workflow dependency optimization, monitoring aggregation, or recommender/data pipeline execution.

First Pass

1. Draw the dependency graph: inputs, intermediate artifacts, outputs, and statistics.
2. Identify keys and ordering rules: joins, event IDs, timestamps, artifact dependencies, data snapshots, and split boundaries.
3. Make every output deterministic: file path, schema, row ordering, rounding, and JSON key names.
4. Compute validation statistics from the produced artifacts, not from raw input assumptions.

Implementation Patterns

Analytical Dataset Assembly

For DuckDB or SQL-based assembly, load each source file as a table, write an explicit query, and export both rows and a summary. Use the schema from the task's inputs; the snippet below is a generic shape, not the exact query.

import duckdb
import json

con = duckdb.connect()
# Load each input CSV as a table; substitute the real names from the task.
for name in ["table_a", "table_b", "fact", "dim"]:
    con.execute(
        f"CREATE TABLE {name} AS SELECT * FROM read_csv_auto('{name}.csv')"
    )

# Pattern: LEFT JOIN to preserve the base side, INNER JOIN for required
# lookups, LEFT JOIN again for optional enrichment, then filter on a status
# flag and project derived columns.
query = """
SELECT
  a.*,
  f.<key>, f.<measure>, f.<rate>, f.<flag_column>,
  CAST(f.<event_ts> AS DATE) AS <event_date>,
  d.<attr_1>, d.<attr_2>,
  -- Derived columns: apply a percentage-style adjustment with 100.0
  -- to keep float arithmetic, e.g. <value> * (1 - <rate>/100.0) AS <derived>.
  f.<measure> * (1 - f.<rate> / 100.0) AS <derived>
FROM table_a a
LEFT JOIN fact f      ON a.<a_key> = f.<a_key>
JOIN      dim  d      ON f.<d_key> = d.<d_key>
LEFT JOIN table_b b   ON f.<f_key> = b.<f_key>
WHERE f.<flag_column> = '<active_value>'   -- filter by status flag
"""
df = con.execute(query).fetchdf()
df.to_csv("<output>.csv", index=False)

Use 100.0 (not 100) for percentage arithmetic so the result stays in floats. Prefer LEFT JOIN whenever a missing related record must not drop the base row; reserve INNER JOIN for genuinely required lookups.

Stream Replay, Deduplication, and Windows

Apply CDC in timestamp order, deduplicate clickstream by event_id before window stats, and compute lateness from event time versus arrival time.

clicks = clicks.drop_duplicates(subset=["event_id"], keep="first").copy()
clicks["ts"] = pd.to_datetime(clicks["timestamp"])
clicks["arr"] = pd.to_datetime(clicks["arrival_time"])
clicks["window"] = clicks["ts"].dt.floor("h")
window_end = clicks["window"] + pd.Timedelta(hours=1)
clicks["late"] = clicks["arr"] > (window_end + pd.Timedelta(hours=1))

For output Parquet, normalize timestamp columns if tests expect strings rather than Arrow timestamp metadata.

CI Workflow Optimization

Infer needs: from real artifact/data dependencies, not from the original sequential order. Keep required fan-in edges.

optimized_duration = critical_path_duration(dependency_graph, job_minutes)

Report every change: parallelized job, cache added, checkout removed, and the final dependency graph.

Monitoring and Latency Pipelines

For synthetic repository or service measurements, preserve sample timestamps, group by service/repo/window, compute percentiles deterministically, and keep raw-to-aggregate traceability.

Validation

• Re-read every output file and verify schema, row count, ordering, and summary numbers.
• Check join cardinality and missing-key behavior.
• Confirm dedup and late-event counts from output data.
• For workflows, ensure all jobs remain and dependency graph is acyclic.
• Compare produced stats against independent calculations.

Common Failures

• Computing reports from intended changes instead of saved outputs.
• Dropping rows through an unintended inner join.
• Deduplicating after aggregating.
• Treating sequential workflow order as a true dependency graph.
• Producing nondeterministic row order that makes tests flaky.