Data Integration End-to-End Workflow
Before you build or debug a data integration pipeline, learn how every pipeline on the C3 Agentic AI Platform is put together. The platform uses the same architecture for files, database tables, and event streams. The architecture is also the same wherever you build the pipeline. You can build a pipeline on the Data Fusion canvas, the visual pipeline editor in C3 AI Studio, or in code from the C3 AI Console.
Knowing the architecture tells you which component to change when a source changes, and where to look when a run fails. Every pipeline has five components that you declare and one of four integration techniques that you choose. Each run follows the same stages and writes status records that you check afterward. Unstructured data shares only the first components and then takes a different path. For an introduction to data integration and the source families (file, database, and streaming), see Data Integration Overview and Structured Data Integration Overview.
The five components of a pipeline
A pipeline is a chain of five components, plus an optional canonical Type when several sources feed one target. Each component handles one concern. When the source changes, you change the component that models the source, and the rest of the chain stays the same. The five components are:
- Source system: The connection to the external storage, such as an Amazon S3 bucket, a Snowflake database, or a Kafka cluster. A SourceSystem holds the connection address and the credentials. You set credentials in the connector wizard or with
setCredentialsat runtime, so they never appear in package code. One source system can serve many source collections. - Source collection: The binding of one dataset, such as a folder, a table, or a topic, to one schema. A SourceCollection also controls how the pipeline runs. The source collection configuration sets chunking, error strictness, archiving, and parse formats. The source collection checkpoint, a saved position in the source, records what the pipeline already loaded.
- Source Type: The schema of one incoming row or message, with the fields exactly as they arrive. The transform reshapes the incoming fields into the form you want to store.
- Transform: The mapping from source fields to target fields. A Transform is metadata that maps each field of one Type to a field of another. A transform maps the fields either as a projection (a field-to-expression map) or as a transformer function. Each mapping step in a chain is a hop. The platform chains transforms automatically and reports a status for each hop.
- Target entity Type: The Type that stores the result. Your application reads the target entity Type with
fetch()andforId(). The transform'sidmapping decides whether reprocessed rows update existing records or create new ones. The target collection writes the records into the entity Type. The target collection configuration, DbTargetCollection.Config, sets whether updates merge or replace fields and whether loads only insert.
A canonical Type is a shared intermediate model between the Source Type and the target. When several sources with different schemas feed one target, each source transforms into a shared Canonical, and one transform maps the canonical to the target. A canonical can be the input of a transform, but it stores no records of its own.
The components chain in a fixed order. Data enters through the source system, is read through the source collection, matches the Source Type, and is reshaped by transforms into the target entity Type:
For worked examples of a canonical pipeline, see Combine Multiple Sources with a Canonical and Combine Multiple Sources with a Canonical via UI.
Choose ingest, incremental ingest, streaming, or virtualization
The same components support four integration techniques. The techniques differ in whether the platform copies the data, when the data moves, and how much moves in each run. Start from your data and how current it must be:
| Your situation | Technique | Data is copied? |
|---|---|---|
| Files or tables the application must own and query at speed | Ingest | Yes |
| A large table where re-reading everything each run is wasteful | Incremental ingest | Yes, changes only |
| Events that must be visible within seconds of being produced | Streaming | Yes, per message |
| A database table you want to query without keeping a copy | Virtualization | No |
You can combine techniques. A common production pattern is an initial bulk ingest followed by scheduled incremental runs. A common development pattern is to virtualize a table while you prototype, then switch to ingest when the application needs its own copy.
Ingest
Ingest copies data through a transform into an entity Type that the application owns. Ingest is the default technique for files on cloud storage and for database tables. REST API sources also ingest, through an ApiSourceCollection. The platform reads the source, reshapes each row, and stores the result, so application queries run against platform data, independent of the source.
Each run moves data in bulk, which makes ingest a batch technique. The data is as current as the last run, and the copy uses platform storage. Ingest fits when the application queries the data heavily, when the source cannot handle application query load, or when the data needs reshaping before use.
For worked examples, see Ingest Files from Amazon S3 Programmatically and Ingest a Snowflake Table Programmatically.
Incremental ingest
Incremental ingest is ingest that resumes where the last run stopped. The source collection's checkpoint records how far each run read, and the next run reads only what changed since then. A daily run against a billion-row table moves only the rows that changed since the day before.
How the platform tracks change depends on the source family:
- SQL and API sources: Change data capture (CDC) is the mechanism. The checkpoint is a cursor value in a field you name in
cdcOrder, such as a last-modified timestamp. A SQL collection reads the changes itself. A REST API collection needs read logic you supply incdcSourcesOverride, and staging turned on with a stage file source collection. - File sources: The platform tracks which files it already processed. For a file store that supports change streams, the checkpoint can be a delta token. A delta token is a marker the store issues so the next read returns only changed files.
- Streaming sources: The platform stores a checkpoint for each partition of the stream and resumes after the last message it processed.
Use incremental ingest when the source is large and mostly stable. For small reference tables that are cheaper to re-read than to track, a full copy on every run works. To re-read a whole SQL source, set fullSync on the collection, which ignores the checkpoint. For the setup for each source family, see Load Data Incrementally.
Streaming
Streaming processes each message as it arrives. The pipeline stays on: a message lands on the topic, flows through the transform, and the platform stores it, typically within seconds. Supported sources include Apache Kafka, Amazon Kinesis, Amazon SQS, Azure Event Hubs, and Google Cloud Pub/Sub.
Streaming fits telemetry, transactions, and monitoring feeds, which lose value while they wait for a scheduled run. Transforms run per message instead of per batch. The collection listens for messages all the time, so you start and stop a streaming pipeline instead of triggering each run. For a worked example, see Ingest a Kafka Message Stream Programmatically.
Virtualization
Virtualization queries the source table live instead of copying it. A SQL source collection in Virtual Table mode exposes the table as an entity Type that mixes External. Each fetch() against that Type runs as a query against the source database. No transform runs, the data is always current, and virtualization uses no platform storage. Mixing External alone does not make a Type virtual: the same Type can feed transforms when the collection loads data.
Virtualization makes the application depend on the source database. Query latency follows the source, every application read becomes load on the source, and the application has no copy if the source is unavailable. Virtualization fits tables that another team owns and updates, and data too large or changing too fast to copy. Virtualization also fits cases where the data must always be current. Virtualization applies to SQL sources and to virtual connectors, such as REST connectors, that query the source live. Virtual REST connectors support live reads; aggregations work only when the connector implements them. For a worked example, see Query a Snowflake Table Programmatically with Virtualization.
How a run executes
For techniques that copy data, a run reads the source, applies the transforms, and stores the result. For file collections, the run has two phases, and keeping them apart helps you find where a pipeline fails:
- Sync registers what exists at the source. Each file becomes a SourceFile record. Sync writes nothing to the target.
- Process moves the data. The platform reads the registered files, applies the transform chain, and stores the result in the target entity Type.
A SQL collection reads rows from the source, starting at the checkpoint when CDC is set up. By default, the same run hands the rows to processing. If you turn on staging for the collection, the platform first writes the rows to files and then processes those files. A streaming collection processes each message as it arrives.
To process large files in parallel, the platform splits each file into file chunks and processes each chunk as an independent batch. The source collection configuration controls chunking. Files under a size threshold process as one unit, and you can cap each chunk by record count or by size. Each chunk has its own status record, so a failure shows at the chunk where it happened.
File chunking splits a large file into batches for parallel processing. File chunking is unrelated to the Chunker stage in unstructured pipelines, which splits document text into passages for embedding.
Each stage of a run writes its own status record, so you can find the exact stage where a failure happened:
Batch and streaming work use the same components; only the trigger differs. A pipeline that copies data runs in one of three execution types: manual, on a schedule, or on arrival of new data. A virtualized Type has no execution type, because each query reads the source directly. To choose an execution type, see Select a Pipeline Execution Type. For the configuration values that control chunking, strictness, and error handling, see Configure a Source Collection.
How you check a run
Every stage of a run writes a status record, and the records form a hierarchy you can drill into. All status records extend DataIntegStatus, so you query them the same way:
| Status Type | One record per | Tells you |
|---|---|---|
| SourceStatus | Source file or sync batch | Whether the source content was read, and how many rows succeeded or failed |
| SourceChunkStatus | Chunk of a file | Which part of a file failed, without re-reading the whole file |
| TransformStatus | Transform hop | Whether the mapping applied cleanly, for each transform in the chain |
| TargetStatus | Target write | Whether the write to the entity Type succeeded |
To find where a failed load stopped, check the records in pipeline order:
- Check the
SourceStatusrecord for the file or batch. - Check the
SourceChunkStatusrecord for the failed chunk. - Check the
TransformStatusrecord for each hop. - Check the
TargetStatusrecord for the write.
A run is clean when every level shows the state completed and a failed count of 0.
A clean status shows that the run executed. A clean status does not show that the data is right. By default, the pipeline drops source columns that the Source Type does not define, and the run completes with zero errors. Turn on strictness in the source collection configuration to fail on unexpected fields. Use validation rules to check the stored records after the load.
Status records accumulate with every run until you remove them. Plan a retention cleanup for production applications.
To monitor runs and drill into failures in C3 AI Studio, see Monitor Your Data Pipeline. The "Check a run from the Console" section of Monitor Your Data Pipeline also shows how to count status records before a cleanup. To check the loaded records against rules, see Validate Your Data Pipeline. To save the configuration of a canvas pipeline to your package so it deploys with the package, see Promote and Version a Data Fusion Pipeline.
Where unstructured data takes a different path
Unstructured Data Integration (UDI) uses the same first steps as structured pipelines: source systems, source collections, and file sync. After sync, documents go through parsing, text chunking, enrichment, and embedding stages for semantic search and retrieval, instead of a transform into an entity Type:
Because the first steps are shared, UDI setup starts with the same connection, credential, and sync concepts. For the unstructured stages, see Unstructured Data Pipeline Architecture.
See also
- Data Integration Overview
- Structured Data Integration Overview
- Integrate Data Programmatically
- Select a Pipeline Execution Type
- Load Data Incrementally
- Configure a Source Collection
- Monitor Your Data Pipeline
- Validate Your Data Pipeline
- Promote and Version a Data Fusion Pipeline
- Unstructured Data Pipeline Architecture