Copilot instructions

1---
2applyTo: "**/pipelines/**/*.py, **/jobs/**/*.py, **/dags/**/*.py, **/consumers/**/*.py, **/producers/**/*.py, **/models/**/*.sql, **/dbt/**/*.sql"
3---
4 
5## Context
6 
7This instruction file applies to data pipeline code: Kafka producers and consumers, Apache Spark PySpark jobs, dbt SQL models, and Airflow / AWS Step Functions DAG definitions. The guiding principle is **idempotency** — every pipeline step must be safe to re-run without producing duplicate or corrupted output. All pipeline code must emit observability metrics and route unprocessable records to a Dead-Letter Queue (DLQ).
8 
9---
10 
11## Coding Standards
12 
13- **Idempotency first:** Every step must use UPSERT/MERGE, partition overwrite, or deduplication keys — never plain INSERT
14- **Manual Kafka commit:** `enable.auto.commit=false`; commit only after successful downstream write
15- **Dead-letter everything:** Unprocessable records go to DLQ with full metadata — never silently drop
16- **Schema contracts:** All data crossing a system boundary has a registered, versioned Avro/JSON Schema Registry schema
17- **`logging` not `print()`:** `logging.getLogger(__name__)` in every module; structured fields
18- **DataFrame API only:** PySpark DataFrame API exclusively in new code — never RDD API
19- **No business logic in dbt staging:** Staging models rename columns only; transformations go in intermediate layer
20- **Observability metrics:** Emit `records_read`, `records_written`, `records_failed`, `duration_seconds` for every job run
21 
22---
23 
24## Preferred Patterns
25 
26### Kafka Producer (idempotent, Schema Registry)
27 
28```python
29# ✅ CORRECT — idempotent producer with Avro serialisation
30from confluent_kafka import Producer
31from confluent_kafka.schema_registry import SchemaRegistryClient
32from confluent_kafka.schema_registry.avro import AvroSerializer
33 
34registry = SchemaRegistryClient({"url": settings.schema_registry_url})
35serialiser = AvroSerializer(registry, ORDER_AVRO_SCHEMA)
36 
37producer = Producer({
38    "bootstrap.servers": settings.kafka_bootstrap_servers,
39    "acks": "all",
40    "enable.idempotence": True,
41    "max.in.flight.requests.per.connection": 5,
42})
43 
44def publish_order_event(event: OrderPlacedEvent) -> None:
45    producer.produce(
46        topic="orders.placed",
47        key=event.order_id,
48        value=serialiser(event.to_dict(), SerializationContext("orders.placed", MessageField.VALUE)),
49        on_delivery=_delivery_callback,
50    )
51    producer.flush()
52```
53 
54### Kafka Consumer (manual commit, DLQ)
55 
56```python
57# ✅ CORRECT — manual commit, DLQ routing
58consumer = Consumer({
59    "bootstrap.servers": settings.kafka_bootstrap_servers,
60    "group.id": "order-processor",
61    "auto.offset.reset": "earliest",
62    "enable.auto.commit": False,
63})
64 
65while True:
66    msg = consumer.poll(timeout=1.0)
67    if msg is None:
68        continue
69    try:
70        event = deserialise(msg.value())
71        processor.handle(event)
72        consumer.commit(message=msg)
73    except ProcessingError as exc:
74        logger.error("Processing failed; routing to DLQ", exc_info=True)
75        dlq_producer.produce(topic="orders.placed.dlq", value=msg.value(), headers=_error_headers(exc))
76        consumer.commit(message=msg)
77 
78# ❌ WRONG — auto-commit loses records on failure
79Consumer({"enable.auto.commit": True})
80```
81 
82### PySpark Job (DataFrame API, explicit schema)
83 
84```python
85# ✅ CORRECT — explicit schema, partition overwrite, observability
86from pyspark.sql import SparkSession
87from pyspark.sql.types import StructType, StructField, StringType, TimestampType
88 
89ORDER_SCHEMA = StructType([
90    StructField("order_id", StringType(), nullable=False),
91    StructField("customer_id", StringType(), nullable=False),
92    StructField("created_at", TimestampType(), nullable=False),
93])
94 
95spark = SparkSession.builder.appName("OrderEnrichmentJob").getOrCreate()
96 
97raw_df = spark.read.schema(ORDER_SCHEMA).parquet(input_path)
98enriched_df = raw_df.join(customer_df, on="customer_id", how="left")
99enriched_df.write.mode("overwrite").partitionBy("created_date").parquet(output_path)
100 
101logger.info("records_read=%d records_written=%d", raw_df.count(), enriched_df.count())
102 
103# ❌ WRONG — inferred schema, RDD API, no observability
104rdd = spark.sparkContext.textFile(input_path)
105```
106 
107### dbt Layered Models
108 
109```sql
110-- ✅ CORRECT: staging model — rename only, no business logic
111-- models/staging/stg_orders.sql
112select
113    order_id,
114    cust_id       as customer_id,
115    order_ts      as placed_at,
116    total_amt_gbp as total_amount_gbp
117from {{ source('raw', 'orders') }}
118 
119-- ✅ CORRECT: intermediate model — business logic here
120-- models/intermediate/int_orders_with_status.sql
121select
122    o.order_id,
123    o.customer_id,
124    o.placed_at,
125    o.total_amount_gbp,
126    case when s.fulfilled_at is not null then 'FULFILLED' else 'PENDING' end as status
127from {{ ref('stg_orders') }} o
128left join {{ ref('stg_fulfilments') }} s using (order_id)
129 
130-- ❌ WRONG: business logic in staging model
131select order_id,
132    case when total_amt_gbp > 1000 then 'HIGH_VALUE' else 'STANDARD' end as tier
133from raw.orders
134```
135 
136---
137 
138## Anti-Patterns — Do NOT Generate
139 
140```python
141# WRONG: auto-commit on Kafka consumer — loses messages on crash [BLOCKER]
142Consumer({"enable.auto.commit": True})
143 
144# WRONG: silently dropping failed records [BLOCKER]
145try:
146    process(record)
147except Exception:
148    pass  # record silently lost
149 
150# WRONG: RDD API in new PySpark code [MAJOR]
151rdd = sc.textFile(path).map(lambda x: x.split(","))
152 
153# WRONG: inferred schema in Spark — breaks on empty partitions [MAJOR]
154df = spark.read.parquet(path)  # no explicit schema
155 
156# WRONG: plain INSERT — not idempotent [MAJOR]
157cursor.execute("INSERT INTO orders VALUES (%s, %s)", (order_id, amount))
158 
159# WRONG: print() in pipeline code [MAJOR]
160print(f"Processed {count} records")
161 
162# WRONG: business logic in dbt staging model [MINOR]
163-- stg_orders.sql
164select *, total_amt_gbp * 0.2 as vat_amount from raw.orders
165```
166 
167---
168 
169## Dependencies & Versions
170 
171| Technology | Version | Notes |
172|-----------|---------|-------|
173| confluent-kafka | 2.x | `acks=all`, `enable.idempotence=True` for producers |
174| apache-spark (PySpark) | 3.5+ | DataFrame API only; use `SparkSession.builder` |
175| dbt-core | 1.7+ | `staging → intermediate → mart` layer convention |
176| great-expectations | 0.18+ | Data quality suite; `checkpoint.run()` in CI |
177| apache-airflow | 2.8+ | Task-level idempotency; use `execution_date` as partition key |
178| boto3 | 1.34+ | AWS Step Functions, S3 access; use `waiters` for polling |
179 
180---
181 
182## Test Conventions
183 
184- Unit test pipeline functions with mocked Kafka client (`MagicMock`) and in-memory DataFrames
185- Integration test Kafka consumers with `testcontainers-python` Kafka container
186- Test idempotency: run the pipeline twice on the same input; verify output row count does not double
187- Test DLQ routing: inject a record that will fail processing and verify it appears in the DLQ topic
188- For dbt, use `dbt test` with `not_null`, `unique`, `relationships` tests on every mart model
189- Verify observability metrics are emitted: assert `records_read`, `records_written` counters in test output
190