jk-shah opened a new issue, #41223:
URL: https://github.com/apache/beam/issues/41223
### Description
In production data pipelines (e.g. Bronze -> Silver -> Gold), validating
records against declarative data quality rules is a fundamental requirement.
Currently in Apache Beam, implementing validation requires writing custom
multi-output DoFns and branching logic. In Beam YAML, there is no declarative
way to specify data quality rules with automated quarantine actions.
### Proposed Architecture
Add an `Expect` Schema-Aware PTransform in `apache_beam.transforms.expect`
(and register `type: Expect` in Beam YAML):
```yaml
pipeline:
transforms:
- type: ReadFromCsv
config: {path: 'gs://bucket/raw/*.csv'}
- type: Expect
config:
rules:
- name: positive_amount
expression: 'amount > 0.0'
on_violation: dlq
- name: non_null_customer
expression: 'customer_id IS NOT NULL'
on_violation: fail
- name: valid_email
expression: "email LIKE '%@%'"
on_violation: drop
dlq_output: InvalidRecords
- type: WriteToParquet
input: Expect
config: {path: 'gs://bucket/silver/'}
- type: WriteToJson
input: Expect.InvalidRecords
config: {path: 'gs://bucket/dlq/'}
```
### Key Capabilities
1. **Tagged Outputs**: Emits `Tag("main")` for passed records and
`Tag("dlq")` for quarantined records.
2. **Standard Beam Metrics**: Emits
`beam:metric:user:expectation_passed:{rule}` and
`expectation_quarantined:{rule}` for real-time monitoring on Dataflow/Flink.
3. **Pluggable Violation Actions**: `dlq`, `drop`, or `fail` (aborts worker).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
