Integrate Data Programmatically

You can build a data integration pipeline on the C3 Agentic AI Platform on the Data Fusion canvas or in code. The canvas is the visual pipeline editor in C3 AI Studio. In code, you run commands in the C3 AI Console, the browser console for your application. You also write package source in your IDE, which the platform syncs to the application. Build in code when you want a pipeline you can review like any other code, reproduce in another environment, and run from a script.

Both routes create the same Types and metadata. Source systems and source collections that you create from the Console appear on the Data Fusion canvas. You can also inspect canvas-built objects from the Console, and change them while the application is in dev mode. A programmatic pipeline has five components, which you define in your package or the Console. You start a pipeline's runs with one of three triggers, and code gives you settings beyond the ones on the canvas. Five quickstarts each build one pipeline end to end.

For an introduction to data integration, see Data Integration Overview. For the pipeline components and run model that both routes share, see Data Integration End-to-End Workflow.

Copy link to this section

Pipeline components

Every pipeline that copies data into your application has the same five components, whatever the source:

• Source system: The connection to the external system, such as an Amazon S3 bucket or a Kafka cluster, and the home for its credentials.
• Source collection: The binding that points one dataset (a folder, a table, or a topic) at one schema.
• Source Type: The schema of a single incoming row or message.
• Transform: The mapping from source fields to target fields, defined by a Transform.
• Target entity Type: The stored result that your application queries. A target collection writes the transformed rows into it.

Virtualization reads a database table in place instead of copying it. Your application fetches from an entity Type that mixes External. Each fetch queries the source table directly, so no transform runs and the platform copies no records. An entity Type that mixes External can also feed a transform when you load the table instead of querying it in place.

When several sources with different schemas feed one target, a canonical Type sits between them. Each source has its own transform into the canonical Type, and one transform maps the canonical Type to the target. A canonical Type mixes Canonical, which makes it a Source Type that holds no data of its own. The canonical pattern works with any kind of source that copies data: files, SQL tables, and streams. In one run, the platform applies every transform in the chain, from source to canonical and from canonical to target.

Copy link to this section

Filter and compute values in a transform

Most transforms map fields with a projection. A projection's keys are target field names, and each value is a C3 expression over the source fields, written as a string. A nested object maps a reference field on the target. When expressions can compute every field, use a projection. This file, saved as metadata/Transform/PersonSource-PersonTarget.json, maps three fields:

{
  "name": "PersonSource-PersonTarget",
  "source": "PersonSource",
  "target": "PersonTarget",
  "projection": {
    "id": "concat('person_', personName)",
    "personName": "personName",
    "nameUpper": "upperCase(personName)"
  }
}

A transform can also skip rows and compute values in code:

• condition: an expression that each source row must meet. The transform skips rows that do not meet it, and the run counts them as skipped on the transform status, not as failed.
• transformer: a function that receives the source object and returns the target object, in place of a projection. To call an expression function from a transformer, pass the expression and the source object to Expr.eval. Field names in the expression resolve against the source object.

This transform file, saved as metadata/Transform/PersonSource-PersonTarget.js, keeps only rows from California and builds two fields with expression functions:

data = {
  name: 'PersonSource-PersonTarget',
  source: 'PersonSource',
  target: 'PersonTarget',
  condition: stateAbbr == "CA",
  transformer: function (srcObj) {
    return PersonTarget.make({
      id: 'person_' + srcObj.personName,
      personName: srcObj.personName,
      stateAbbr: srcObj.stateAbbr,
      nameUpper: Expr.eval('upperCase(personName)', srcObj),
      cityLabel: Expr.eval('concat(city, ", ", stateAbbr)', srcObj),
    });
  },
};

To test a transform from the C3 AI Console before you run the pipeline, read a few rows from the source collection and preview the transform. The preview stores nothing:

var rows = SourceCollection.forName('peopleColl').readObjs(ObjsOperSpec.make({ limit: 10 })).collect();
PersonSource.transformSource(rows);

Replace peopleColl with the name of the source collection that feeds PersonSource.

The result lists the transformed objects by target Type, and its statistics count the rows that the condition skipped. To run one transform on one object, get the Transform with forName and pass the object to its transform method. The method returns a stream, so add .collect():

Transform.forName('PersonSource-PersonTarget')
  .transform(PersonSource.make({ personName: 'Eli', stateAbbr: 'CA', city: 'Davis' }))
  .collect();

transformSource previews only the transforms that start at the Type you call it on. For a chain through a canonical Type, call it again on the canonical Type.

Copy link to this section

Where you define each component

Types and transforms live in your package, source systems and collections live in the Console or the package, and credentials never go in the package:

For the credential fields each source accepts, see Authenticate File Sources, Authenticate Database Sources, and Authenticate Streaming Sources. For the role you need to build pipelines in code, see the "Roles for programmatic data integration" section of Data Integration Roles and Permissions.

Settings that you tune on the Data Fusion canvas belong to the environment where you set them. To promote canvas settings to other environments, save the settings to your package, as described in Promote and Version a Data Fusion Pipeline.

Copy link to this section

Ways to run a pipeline

A pipeline that copies data runs in one of three ways, and the source family decides which ones apply. A source family is the kind of source a collection reads: files, SQL tables, API endpoints, or message streams.

Only startOnScheduleMode() or a CronJob creates a schedule. Setting processMode and processSchedule in the collection configuration records the cadence but starts nothing. An API source collection runs on a schedule only when it sets cdcSourcesOverride and stages its data.

Virtualization has no run. Each fetch on the entity Type that mixes External queries the source when your application reads it.

To choose a trigger and set it up step by step, see Select a Pipeline Execution Type.

Copy link to this section

What you control in code

Data Fusion exposes the settings most pipelines need. The Edit properties dialog of a source collection sets chunk sizes, error thresholds, retry counts, and archiving. The Edit properties dialog also sets the integration schedule and the CDC order fields. CDC loads only the rows that changed since the last run. The Entity Type node sets the persistence mode, merge or replace, and staged updates.

In code, you can also control these settings:

• Every configuration field: The source collection configuration has more than 30 fields, including processing priority.
• Insert-only loads: Set createOnly on the target collection with DbTargetCollection.Config, so a load creates records and never updates existing ones.
• Jobs before and after a load: Run BatchJobs, platform jobs that process data in batches, before and after a load.
• Checkpoint inspection: Read the full checkpoint record, which shows how far the last incremental run read, with cdcCheckpoint(). On the canvas, the Execute Pipeline dialog shows only the last value, as LAST UPDATED.
• Automation: Run the full lifecycle from a script (connect, model, run, verify). You can then test a pipeline in continuous integration (CI) and rebuild it in a new environment. For example, the Kafka quickstart sends test messages to its topic from code.
• API sources: Connect a REST API through an ApiSourceCollection. Data Fusion does not draw API source collections on the canvas.

These pages cover configuration, triggers, incremental loads, and BatchJobs in detail:

• Configure a Source Collection
• Select a Pipeline Execution Type
• Load Data Incrementally
• Run BatchJobs Before and After a Pipeline

Copy link to this section

Choose a quickstart for files, tables, streams, or multiple sources

Find the row that matches your data, then follow that quickstart:

The Amazon S3 and Snowflake ingest quickstarts follow six steps: connect, inspect the source, model, create the collection, run, and verify. The other three quickstarts adapt the six steps to their source. The virtualization quickstart ends by querying the table live. The Kafka quickstart creates the collection before it sets credentials. The canonical quickstart stages its own sample files, so it has no connect step. Every quickstart ends with a table of common problems and their fixes.

Copy link to this section

See also

• Data Integration Overview
• Configure File Sources
• Configure SQL Sources
• Configure Streaming Sources
• Monitor Your Data Pipeline