Structured Data Integration Overview
Your application needs data that lives in databases, files, or event streams. Structured Data Integration (SDI) brings that data into your application's entity Types as records you can query. An entity Type is a C3 AI Type whose records the C3 Agentic AI Platform stores. For example, an Order Type holds one record per order. SDI handles data with a fixed schema, which means every row has the same named columns. Such data includes database tables, CSV, JSON, Parquet, Avro, and ORC files, and event-stream topics.
Before you build your first structured pipeline, learn the three stages every pipeline goes through and the Types you configure. Then choose the source family that fits your data and open the matching setup page. A source family is the kind of source a collection reads: files, SQL databases, or event streams.
For documents and other unstructured content, see Understand Unstructured Data Integration Pipelines.
The pipeline lifecycle
Every structured pipeline moves through the same three stages:
- Configure: Declare the source system, the source collection, the Source Type that describes one incoming row, and the transform that maps each row to your entity Type. You configure a pipeline once, then revise it when the source changes.
- Run: Start a run to load data. The platform reads the source, applies your transform, and stores the result. You can run a pipeline manually, on a schedule, or when new data arrives. Which triggers a source collection supports depends on its source family. For large inputs, the platform splits the data into chunks and processes the chunks in parallel, so one run never loads every record at once.
- Monitor and update: Check the status of each run, find the cause of a failure, and update the data model as the source evolves. Data validation rules check the stored records for quality problems before other processes use them. The platform offers foreign-key rules and custom (Lambda) rules.
For every run, the platform writes status records that let you find the exact step that failed. Each status record is a DataIntegStatus, and each kind covers one step:
- SourceStatus: Reading the source. A SourceChunkStatus record covers each chunk.
- TransformStatus: Applying a transform.
- TargetStatus: Writing the records to the target.
To set up run triggers, see Select a Pipeline Execution Type. To check runs and validate the stored records, see Monitor Your Data Pipeline and Validate Your Data Pipeline. For the components, the run model, and the status records in more depth, see Data Integration End-to-End Workflow.
Types in a structured pipeline
Every structured pipeline uses these Types:
| Type | Role |
|---|---|
| SourceSystem | The external system that holds the data, such as an Amazon S3 bucket, a Snowflake database, or a Kafka cluster. The source system also holds the connection details and credentials. |
| SourceCollection | One dataset inside that system, such as a folder, a table, or a topic. |
| Source | The base for your Source Type, which describes the columns of one incoming row as it arrives. |
| Canonical | Optional. A shared intermediate shape that several sources map into, so one transform to your entity Type serves all of them. A canonical Type is also a Source Type, so it can feed another transform. |
| Transform | Metadata that maps the fields of a Source Type to a target Type. A transform uses a projection, which maps each target field to an expression on the source fields, or a transformer, a JavaScript function. |
| TargetCollection | The stage that writes the transformed records. DbTargetCollection writes to the application's datastore. Other target collections write to files, a data lake, a message stream, or a virtual target. |
| Your entity Type | The final destination. Your application reads the records with fetch() and forId(). |
A pipeline with one source uses one transform, from the Source Type to your entity Type. When several sources with different schemas feed one entity Type, each source transforms into a shared canonical Type. One more transform then maps the canonical Type to your entity Type. The platform runs every transform in the chain for each record. In code, calling process() on a Source record runs the whole chain.
The diagram shows both paths, from the source system to the application datastore:
To browse your entity Types and their relationships, open the Object Model tab of Data Fusion. For a worked example with a canonical Type, see Combine Multiple Sources with a Canonical via UI or, in code, Combine Multiple Sources with a Canonical.
Choose a file, SQL, or streaming source family
Choose a source family by where your data lives. Each family uses its own source system and source collection Types, and the rest of the pipeline works the same way:
- File sources: Files on cloud storage, read through a FileSourceSystem and a FileSourceCollection. The Data Fusion connectors cover Amazon S3, Google Cloud Storage, Azure Blob Storage, OneDrive and SharePoint, and Google Drive. You can also upload files from the file explorer in Data Fusion.
- SQL sources: Databases, read through a SqlSourceSystem and a SqlSourceCollection. A SQL source collection runs in one of two modes. Virtual Table queries the table in place each time your application reads it, without copying the data. Load Data copies rows into your application with change data capture (CDC), which reads only the rows that changed since the last run.
- Streaming sources: Continuous event streams, read through a CloudMessageSourceSystem and a CloudMessageSourceCollection. The Data Fusion connectors cover Apache Kafka, Amazon Kinesis, and Azure Event Hubs. In code, the platform also supports Amazon SQS and Google Cloud Pub/Sub.
Connectors for business applications and APIs, such as Box, create a SQL source system, so they follow the SQL pattern. The platform also offers a data lake source.
Choose the family by how you reach the data, not by where it is stored. For example, Delta Lake tables sit in cloud storage, but the Delta Lake connector reads them through a Databricks SQL endpoint. The Delta Lake connector is therefore a SQL connector.
To compare ingest, incremental ingest, streaming, and virtualization before you choose, see Data Integration End-to-End Workflow.
Where to start
Each source family has a setup page and a worked example in Data Fusion that uses one representative connector. The same steps apply to the other connectors in that family:
- Files: Read Configure File Sources, then follow the Amazon S3 example in Connect with File Source System.
- Databases: Read Configure SQL Sources, then follow the Snowflake example in Connect with SQL Source System.
- Event streams: Read Configure Streaming Sources, then follow the Kafka example in Connect with Streaming Source System.
To build the same pipelines with C3 AI Console commands and package files instead of Data Fusion, see Integrate Data Programmatically. To check the roles you need, see Data Integration Roles and Permissions.