C3 AI Documentation Home

Select a Pipeline Execution Type

A data integration pipeline in the C3 Agentic AI Platform has three parts. A source system holds the connection and credentials for an external store. A source collection names the files, table, or topic to read from that store. A transform maps each record onto a target entity Type, the Type that stores the loaded data. These parts define what data moves and how it is shaped. The execution type defines when the pipeline runs, and a pipeline with no execution type moves no data.

Choose an execution type after you build a working pipeline. You then run the pipeline manually one time. Next, you start a schedule (SQL or file) or on-arrival processing (streaming, or files in an Amazon S3 or Google Cloud Storage inbox). Finally, you verify the first scheduled run. Fix common problems lists the errors you are most likely to see and their fixes.

Prerequisites

Make sure you have:

The examples use collections named ordersFileColl (file), ordersDbColl (SQL), and ordersStreamColl (streaming), and a target entity Type named OrderReading. Replace these names with your own.

Run manually, on a schedule, or on arrival

Match the execution type to how the data arrives, not to how often you want to see it. The source family of the collection decides which call starts each type:

Execution typeSource familyCallUse when
ManualFile, SQLSourceFile.syncAll(...) (file) or process() (SQL)Development, one-time loads, backfills, and runbook-driven operations
ScheduledSQLstartOnScheduleMode() on the collectionThe source table changes on a cadence and the collection uses change data capture
ScheduledFileA CronJob you define that calls syncAllFiles land in the inbox on a cadence
On arrivalStreamingstartOnArrivalMode() on the collectionEvents arrive continuously on a message broker
On arrivalFile (Amazon S3 or Google Cloud Storage inbox)startOnArrivalMode() on the collectionEach file should load as soon as it lands in the inbox

Each source collection has a configuration, its stored settings, with two scheduling fields: processMode and processSchedule. Setting these fields records the cadence you want, but it does not start a schedule. Only the calls in the execution type table start a pipeline.

Virtualization reads a remote SQL table live each time you query it, so it copies no data and has nothing to trigger. For the virtualization model, see Query a Snowflake Table Programmatically with Virtualization.

Run a pipeline manually

Run every pipeline manually at least one time before you schedule it. A manual run lets you control the timing, so a broken transform is easier to find and fix than in a scheduled run.

For a file collection, one call registers every new file in the collection's inbox (the folder the collection reads from) and processes it through the transform:

JavaScript
var inbox = FileSourceCollection.forName('ordersFileColl').inboxUrl();
SourceFile.syncAll(inbox, DataIntegSpec.make({ process: true }));

DataIntegSpec holds the options for the run; process: true tells syncAll to process the files it registers, not only record them. syncAll is asynchronous: it queues the files for processing and returns.

Other DataIntegSpec fields override the collection configuration for this run only. For example, this run processes each file as one chunk, even when the collection sets chunkSize:

JavaScript
SourceFile.syncAll(inbox, DataIntegSpec.make({ process: true, doNotChunk: true }));

To register only some files, pass their File objects to SourceFile.syncFileMetadataBatch with the same DataIntegSpec. To sync files that are already registered again, pass their SourceFile records to SourceFile.syncBatch. To process one registered file, call process on its record:

JavaScript
SourceFile.fetch({ filter: "sourceCollectionName=='ordersFileColl'", limit: 1 }).objs[0]
  .process(DataIntegSpec.make({}));

For a SQL collection that uses change data capture (CDC), call process() on the collection. CDC records the last row loaded so that the next run reads only newer rows; Load Data Incrementally shows how to configure it.

JavaScript
SqlSourceCollection.forName('ordersDbColl').process(DataIntegSpec.make({}));

By default, process() on a SQL collection reads the changed rows, splits them into chunks, and advances the checkpoint before it returns. The chunks then transform in the background. To transform the rows before the call returns, pass DataIntegSpec.make({ doNotChunk: true }). When the collection stages data into files first, the staged files process asynchronously, the same way as a file collection.

If a chunk fails after the checkpoint advances, the next run does not read its rows again. To reload the chunk's rows, run the collection in Data Fusion with Reset Run State turned on. The reset clears the checkpoint, so the run reloads the whole table. The reset also clears queued work and run status records for the whole application. For the steps, see Load Data Incrementally.

Confirm the run with SourceStatus, which records the outcome of each run for a collection, and with a row count on the target entity Type:

JavaScript
c3Grid(SourceStatus.fetch({
  filter: "sourceCollectionName=='ordersFileColl'",
  order: 'descending(meta.created)'
}));
OrderReading.fetchCount();

Schedule a SQL pipeline

SQL collections have a built-in schedule because the platform tracks a CDC checkpoint for them: the value of the last row loaded. Set the collection field cdcOrder so that each scheduled run knows which rows to read. A collection that does not stage its data needs cdcOrder even when fullSync is on. Load Data Incrementally describes both fields.

  1. Set the cadence. The schedule is a Quartz cron expression stored in the processSchedule field of the collection's configuration. The default is 0 0 0 * * ?, which is midnight UTC daily. Set the cadence before you start the schedule:

    JavaScript
    var coll = SqlSourceCollection.forName('ordersDbColl');
coll.config().setConfigValue('processSchedule', '0 0 2 * * ?', ConfigOverride.APP);

  2. Start the schedule:

    JavaScript
    coll.startOnScheduleMode();

    startOnScheduleMode() creates and resumes a CronJob named cron_<sourceSystemName>_<collectionName>. The job runs on the processSchedule cadence, and each run processes the collection the same way a manual process() call does.

  3. Confirm that the CronJob exists:

    JavaScript
    CronJob.fetch({ filter: Filter.contains('id', 'ordersDbColl') });

Each scheduled run advances the CDC checkpoint, so it picks up only rows that changed since the previous run. A run against an unchanged table completes and writes nothing, which is the expected result.

To stop the schedule, call stopOnScheduleMode() on the collection. stopOnScheduleMode() pauses the CronJob and returns the collection to manual execution.

startOnScheduleMode() reads processSchedule only when it creates the CronJob. After the CronJob exists, startOnScheduleMode() keeps the schedule of the existing CronJob, so a new processSchedule value alone does not change the cadence. To change the cadence, set the new value, remove the CronJob, and start the schedule again:

JavaScript
var coll = SqlSourceCollection.forName('ordersDbColl');
coll.config().setConfigValue('processSchedule', '0 0 4 * * ?', ConfigOverride.APP);
coll.processCron().remove();
coll.startOnScheduleMode();

Other source collections, such as API collections, can use startOnScheduleMode() only when they stage data. Without staging, the call fails with an error that the collection is not configured for staging.

startOnScheduleMode() does not schedule file collections, because they have no CDC checkpoint to advance. On a file collection, the call throws this error:

Text
FileSourceCollection <collectionName> requires cdcSources or cdcSourcesOverride to be implemented for on-schedule processing.

Schedule a file collection with your own CronJob instead, as described in Schedule a file pipeline with a CronJob.

Schedule a file pipeline with a CronJob

For file collections, you define the CronJob yourself. The job calls the same SourceFile.syncAll action you run manually. Each scheduled run registers the files that landed in the inbox since the last run and processes them.

  1. Define and save the job, then make sure it is active:

    JavaScript
    var inbox = FileSourceCollection.forName('ordersFileColl').inboxUrl();

var job = CronJob.make({
  id: 'cron_di_ordersFileColl',
  name: 'DI sync and process ordersFileColl',
  action: { typeName: 'SourceFile', actionName: 'syncAll' },
  inputs: {
    urlOrEncodedPath: inbox,
    spec: DataIntegSpec.make({ process: true })
  },
  concurrent: false,
  scheduleDef: { cronExpression: '0 0 2 * * ?', skipOverdue: true }
}).upsert();

job.resume();

  2. Check the first scheduled run, as described in Verify the schedule.

The fields in this CronJob decide how the job behaves:

  • inputs: must match the action signature exactly: syncAll(urlOrEncodedPath: string, spec: DataIntegSpec). Pass the spec as DataIntegSpec.make({ process: true }). upsert() does not check the inputs. If the inputs do not match the signature, every run fails, and a spec without process: true registers files without processing them.
  • concurrent: false: prevents a slow run from overlapping the next one. The default is false.
  • skipOverdue: true: when a run completes, the job bases the next trigger time on the current time, not on the completed run's scheduled time. This setting applies only when concurrent is false.
  • resume(): makes sure the job is active. If the job is already active, the call has no effect.

After a run processes a file, the platform moves it from the inbox to the collection's archive folder, and later runs do not register it again. Each run processes only the files that arrived since the previous run.

If you drop a file with the same name into the inbox again, the platform updates the existing source file record and processes the file again. When the transform sets a stable ID on each target row, the new values update the existing rows instead of adding new ones.

To stop the schedule, pause the job:

JavaScript
var job = CronJob.make('cron_di_ordersFileColl').get();
job.pause();

To retire the pipeline, remove the job:

JavaScript
CronJob.make('cron_di_ordersFileColl').get().remove();

Process data on arrival

A streaming collection processes each message as it lands on the broker topic, so you do not schedule it. Set the collection's process mode to on arrival, then call startOnArrivalMode() on the collection:

JavaScript
var coll = CloudMessageSourceCollection.forName('ordersStreamColl');
coll.config().setConfigValue('processMode', 'ON_ARRIVAL', ConfigOverride.APP);
coll.startOnArrivalMode();

startOnArrivalMode() only registers the collection with the broker. Messages reach only collections whose process mode is ON_ARRIVAL, so a collection left in MANUAL mode loads nothing.

To stop processing, call stopOnArrivalMode() on the collection. On a streaming collection, the call unregisters the collection but leaves the process mode at ON_ARRIVAL. To return the collection to manual execution, also set the process mode back:

JavaScript
coll.stopOnArrivalMode();
coll.config().setConfigValue('processMode', 'MANUAL', ConfigOverride.APP);

Messages that arrive while the collection is stopped stay on the topic and load when you start it again, as long as the topic still retains them. Messages that the collection reads while its process mode is MANUAL do not load.

If other source collections read the same topic, set the process mode back to MANUAL when you stop. While a stopped collection stays ON_ARRIVAL, startOnArrivalMode() and stopOnArrivalMode() on the other collections for that topic do nothing.

You can also select Pause on arrival on the collection node in Data Fusion. The collection then shows the process mode Manual. To confirm that messages reach the target, run the SourceStatus query and row count from Run a pipeline manually. In both, use your collection name and target entity Type.

The streaming quickstart, Ingest a Kafka Message Stream Programmatically, covers the full broker setup: the source system, the collection fields, and the credentials. The quickstart creates the source system and the collection before it sets credentials, because credentials reach only the collections that exist when you set them.

A file collection whose inbox is in Amazon S3 or Google Cloud Storage can also process files on arrival. Call startOnArrivalMode() on the file collection; each file then loads after it lands in the inbox, with no syncAll call.

On-arrival processing for files needs bucket event notifications set up for the environment. For Amazon S3, the notifications go to an Amazon Simple Queue Service (SQS) queue. For Google Cloud Storage, the notifications go to a Google Cloud Pub/Sub topic. Without bucket event notifications, startOnArrivalMode() completes and sets the mode to on arrival, but no file loads.

Verify the schedule

Check the first scheduled run in two places: the job's run history and SourceStatus. A job can be active while its runs fail or load nothing.

A CronJob with inputs that do not match the action signature saves without an error and then fails on every run. Always read the run history after the first scheduled run.

The platform stores the run history in JobHistory records, and the CronJob returns them directly:

JavaScript
var job = CronJob.make('cron_di_ordersFileColl').get();
job.lastRuns(5).map(function (r) {
  return r.scheduledTime + ': ' + r.status;
});

A healthy schedule shows COMPLETED entries at the expected cadence. A FAILED entry carries the exact error:

JavaScript
var run = job.lastRun();
run.status;         // "FAILED"
run.error.message;  // for example: "Invalid call to syncAll: ... Invalid value {...} for parameter: spec: DataIntegSpec"

The status field of a JobHistory record tells you whether the job ran and the platform accepted the action. The status field does not tell you whether the rows processed cleanly; SourceStatus records the row outcome, as in a manual run. After the first scheduled run, read lastRuns for the job and run the SourceStatus query from Run a pipeline manually for the data.

Fix common problems

Find the row that matches what you see, then apply the fix:

SymptomCauseFix
The collection configuration has processMode set to ON_SCHEDULE, but nothing runsThe configuration fields record the cadence; they do not start a scheduleCall startOnScheduleMode() (SQL) or create a CronJob (file)
startOnScheduleMode() throws requires cdcSources or cdcSourcesOverrideThe collection is a file collectionDefine your own CronJob, as described in Schedule a file pipeline with a CronJob
startOnScheduleMode() or a scheduled run on a SQL collection fails with incremental (WATERMARK) extraction requires at least one order-by fieldThe collection has no cdcOrder field. A collection that does not stage its data needs cdcOrder even when fullSync is onSet cdcOrder as described in Load Data Incrementally, then start the schedule again
The CronJob runs, but the target row count never changesThe runs fail, most often because inputs do not match the action signatureRead job.lastRuns(5); a FAILED status with error.message names the bad parameter
A scheduled SQL run completes but loads zero rowsThe CDC checkpoint is already at the newest rowExpected when the source is unchanged. To reload, reset the checkpoint as described in Load Data Incrementally
The CronJob exists but never runsThe job is pausedCall resume() on the job

The full error for a SQL collection with no cdcOrder field is:

Text
Validation error: Source collection '<collectionName>': incremental (WATERMARK) extraction requires at least one order-by field (config field `cdcOrder`). Without one the platform cannot derive an incremental filter.

See also

Was this page helpful?