Validate Source Data for Schema Drift
When an upstream system adds, drops, or changes a field, a pipeline that reads it can fail or load wrong values. This change in the shape of the source is schema drift. The C3 Agentic AI Platform can check a source collection for schema drift before or while you load its data. You run the check, review the issues it records, and clear them after you fix the cause.
A source collection is the platform object that reads one feed of files, rows, or messages from a source system. A source collection's Source Type is the C3 AI Type that lists the fields you expect the feed to contain. To set up a source collection, see Configure a Source Collection.
Source data validation and data validation rules
The platform provides two kinds of data validation, and a pipeline that loads reliably uses both:
| Question | Validate the data you ingested | Validate the source data (this page) |
|---|---|---|
| What it inspects | The persisted records in your entity Types | The data a source collection reads, compared with its Source Type |
| What it catches | Broken references, out-of-range values, stale data, business-rule violations | Schema drift: missing fields, extra fields, and fields whose type changed |
| Mechanism | DataValidation.Rule (Foreign Key and Lambda rules) | SourceCollection.validateSourceData, which returns DataIssue.SourceCollection records |
| Where you run it | The Data Validation tab in Data Fusion or code | Code only, from a Jupyter notebook, the C3 AI Console, or your package |
| Guide | Validate Your Data Pipeline | This page |
Data validation rules run on persisted entity Types after ingestion and report failures as DataValidation.Run details. Source data validation runs against the source and reports schema drift as DataIssue.SourceCollection records. Source data validation issues don't appear in the Data Validation tab or on the Data Monitoring page; query them through DataIssue.SourceCollection.
When to use source data validation
Use source data validation when you want to:
- Detect that an upstream source has added, removed, or changed fields compared with the Source Type of your source collection.
- Catch schema drift before a scheduled load, instead of finding it through downstream failures.
- Keep a record of source schema problems, each with a severity, that you can review and resolve.
You can run the check on every kind of source collection, including FileSourceCollection, SqlSourceCollection, and CloudMessageSourceCollection. The shared SourceCollection base Type defines the method.
How source data validation finds schema drift
validateSourceData works in three steps:
- The method infers the current schema of the source. For a file source collection, the method reads a random sample of up to 10 files and combines their fields. For a SQL source collection, the method reads the schema of the table.
- The method compares that schema with the Source Type of the collection.
- The method returns one
DataIssue.SourceCollectionrecord for each mismatch, and stores the records if you ask it to.
The method reports four kinds of mismatch. The method assigns each mismatch a severity:
| Mismatch | Severity |
|---|---|
| The source is missing a required field of the Source Type | CRITICAL |
| A field in the source has a different type | WARNING |
| The source is missing an optional field of the Source Type | MINIMAL |
| The source has a field that the Source Type does not declare | MINIMAL |
For SQL and streaming source collections, the method first checks the connection to the source. The method returns any connection issue, even when you set a severity filter.
The method signature is:
validateSourceData: member function(spec: ValidateSourceDataSpec): [DataIssue.SourceCollection]ValidateSourceDataSpec
Pass a ValidateSourceDataSpec to control which issues the method returns and whether it stores them:
| Field | Type | Purpose |
|---|---|---|
persist | boolean | If true, the method stores the returned issues as DataIssue.SourceCollection records so you can query and count them later. If false or unset, it returns the issues without storing them. |
severity | string enum DataIssueSeverity | If set, the method returns only issues of that severity. With persist set to true, it stores only those issues. Connection issues are returned regardless of this filter. |
If you call validateSourceData() with no spec, the method returns every issue and stores none.
The DataIssue.SourceCollection record
DataIssue.SourceCollection is the issue Type for source collections. The issue Type extends the platform's DataIssue Type. Each record describes one mismatch:
| Field | Meaning |
|---|---|
sourceCollection | The source collection whose source data is out of sync. |
targetType | The Type the issue applies to. For schema drift, this is the name of the collection's Source Type. |
targetField | The field with the issue. Null means that the issue applies to the whole Type. |
targetId | The ID of the affected instance. Null means that the issue applies to all instances of targetType. Schema drift issues always leave this field null. |
errorMsg | A message that describes the drift, for example "Source data has extra field: Nickname". |
severity | One of the DataIssueSeverity levels. The field default is WARNING. |
Severity levels
DataIssueSeverity has three levels. Fetched records show the stored value:
| Level | Stored value | Assigned to |
|---|---|---|
CRITICAL | critical | A missing required field. This drift breaks the load or downstream consumers. |
WARNING | medium | A changed field type. Review it before the next load. |
MINIMAL | low | A missing optional field or an extra field. Record it; it needs no urgent action. |
Run source data validation
The samples on this page are Python. Run them in a Jupyter notebook in your application or in Python code in your package. The same methods are available in JavaScript from the C3 AI Console.
Before you run the samples, you need a source collection whose Source Type is defined, in an application where you can run code.
Fetch a source collection and validate its source data, storing the results. Replace TurbineReadingsCollection with the name of your source collection, and use the Type that matches it, for example FileSourceCollection:
# Fetch the source collection you want to check
sc = c3.SqlSourceCollection.forName('TurbineReadingsCollection')
# Validate incoming source data against the declared schema and store any issues
issues = sc.validateSourceData(
c3.ValidateSourceDataSpec(persist=True)
)
for issue in issues:
print(issue.severity, issue.targetType, issue.targetField, '-', issue.errorMsg)A run with no drift returns an empty list. A run that finds drift prints one line for each issue. In this example output, the collection's Source Type is IrisSource, a sample Source Type for rows of an iris measurement dataset. The source data has an extra Nickname field. Your output shows your own Source Type and fields:
low IrisSource Nickname - Source data has extra field: NicknameRestrict a run to the most serious drift. Pass the stored value critical, not the level name CRITICAL:
critical = sc.validateSourceData(
c3.ValidateSourceDataSpec(persist=True, severity='critical')
)Run without storing when you only want a one-time check:
preview = sc.validateSourceData(c3.ValidateSourceDataSpec(persist=False))To detect drift on a schedule, call validateSourceData from a CronJob or as a step ahead of a scheduled load. Act on any CRITICAL result before the pipeline processes the source.
Query and summarize stored issues
When you validate with persist=True, you can query the stored DataIssue.SourceCollection records.
Fetch the stored issues. The targetType of each issue names the Source Type of the collection it came from:
issues = c3.DataIssue.SourceCollection.fetch().objs
for issue in issues:
print(issue.severity, issue.targetType, issue.targetField, '-', issue.errorMsg)To count issues, call summary with the fields to group by. summary returns DataIssueSummary rows, and each row holds the field values and a count:
# Count issues by severity, then by severity and field
c3.DataIssue.SourceCollection.summary('severity')
c3.DataIssue.SourceCollection.summary('severity, targetField')Resolve issues
Issue records stay in place until you re-check or clear them. After you update the Source Type or fix the upstream source, use one of these methods:
validate(): A member function on a single issue. The method re-checks that issue, removes the record if the drift is resolved, and stores the record again if it is not.validateAll(targetType): Re-checks every issue, or only the issues fortargetType, and removes the ones that are resolved.clearAll(targetType, targetId, targetField): Removes the stored issues for the given scope.targetTypeis required;targetIdandtargetFieldare optional, and leaving one out widens the scope to all IDs or all fields. CallclearAllonDataIssue.SourceCollection, not on the baseDataIssueType.
To refresh the stored issues after a fix, clear the issues for the collection's Source Type, then validate the source data again:
# Clear the stored issues for the collection's Source Type, then store a fresh set
c3.DataIssue.SourceCollection.clearAll('<Source Type>')
sc.validateSourceData(c3.ValidateSourceDataSpec(persist=True))