1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
# CSV → JSONL guarded by a data contract (issue #204).
#
# The `contract:` block is a versioned promise about the pipeline's *output*
# shape: required fields, types, nullability, enum sets, regex patterns, and
# numeric/length bounds. It is enforced per page after transforms and quality
# checks and before the sink write. `on_breach` picks the enforcement policy:
#
# fail — abort the run on the first breach (default; nothing from the
# breaching page is written)
# quarantine — route breaching records to the DLQ, write the rest
# warn — log + count breaches, write everything unchanged
#
# Inspect or publish the contract with the CLI:
# faucet contract cli/examples/csv_to_jsonl_with_contract.yaml
# faucet contract cli/examples/csv_to_jsonl_with_contract.yaml --export json-schema
# faucet contract cli/examples/csv_to_jsonl_with_contract.yaml --export openlineage
#
# Breaching rows are quarantined to ./dlq/contract_breaches.jsonl. Once you fix
# the source data (or the contract), inspect and replay them (issue #281):
# faucet dlq inspect ./dlq/contract_breaches.jsonl
# faucet dlq replay cli/examples/csv_to_jsonl_with_contract.yaml \
# --from ./dlq/contract_breaches.jsonl --dry-run
# faucet dlq discard ./dlq/contract_breaches.jsonl --reason contract --before 7d
version: 1
name: orders_csv_with_contract
pipeline:
source:
type: csv
config:
path: ./orders.csv
contract:
version: "1.0.0"
description: Orders exported for the analytics team.
owner: data-platform
on_breach: quarantine
allow_extra_fields: true
fields:
- name: order_id
type: string
min_length: 1
description: Unique order identifier.
- name: status
type: string
enum: [open, shipped, cancelled]
description: Order lifecycle state.
- name: customer_email
type: string
pattern: '^[^@\s]+@[^@\s]+\.[^@\s]+$'
required: false
nullable: true
dlq:
sink:
type: jsonl
config:
path: ./dlq/contract_breaches.jsonl
sink:
type: jsonl
config:
path: ./orders_out.jsonl