Run BatchJobs Before and After a Pipeline

Your data integration pipeline loads rows from a source collection into a target entity Type through a transform. Some work does not fit a transform. Before a load, you may need to join files or convert them into the format the source collection expects. After a load, target rows may need enrichment, reconciliation, or a refresh of calculated fields.

A BatchJob runs that work. A batch job splits the work into batches, units of work that run in parallel across the cluster, and calls a completion function when every batch succeeds. You can start the pipeline from the completion function.

In this tutorial, you build two jobs. A staging job writes a file to the collection's inbox, the folder that the collection reads new files from, and starts the pipeline. An enrichment job normalizes phone numbers after the load. You also see how to rebuild application dependencies from Data Fusion after a large historical load. You then read the errors of a failed job and fix common problems.

Prerequisites

Before you start, make sure you have:

A package open in your IDE. You save the job Types and their JavaScript implementations in the package source, which the C3 Agentic AI Platform syncs to the application.
A file pipeline that loads an entity Type through a transform. The examples use a Contact Type with a phone field, a file source collection named contactsFileColl, and job Types named ContactStageJob and ContactEnrichJob with type keys CSTGJ and CENRJ. Replace these names with your own. To build a file pipeline you can adapt, see Ingest Files from Amazon S3 Programmatically.
Access to the C3 AI Console of your application. You run the start, status, and diagnosis snippets there.

Understand the batch job contract

A batch job is an entity Type that extends BatchJob and implements three functions that the engine calls in order:

doStart: runs one time. doStart decides what the work is and schedules it in batches with job.scheduleBatch(...).
processBatch: runs one time for each scheduled batch, in parallel, on any node (server) of the cluster.
allComplete: runs one time, only after every batch finished successfully. The return string of allComplete appears in the info field of the job status. If any batch fails, allComplete does not run.

An optional failed function runs when the job's status becomes failed.

BatchJob takes three generic parameters. The first is the job Type itself, the second an options Type (BatchJobOptions unless you define your own), and the third the batch Type. The batch Type is a primitive such as string or a reference Type. Each job Type also declares a type key, a short unique key. All batch jobs share one storage table, and the key tells the platform which Type each stored job belongs to.

Declare each function you implement again with ~ js-server. The ~ re-declares an inherited function, and js-server binds it to your JavaScript implementation file. A function declared with a bare ~ has no implementation language: the Type deploys (provisions) without errors, and then every start fails.

Prepare data and start the pipeline

The staging job prepares source data, then starts the pipeline only after every staging batch succeeds. Start the pipeline from allComplete: it runs only when every batch succeeded, so the pipeline never starts on a half-staged load.

Save the Type as ContactStageJob.c3typ in your package:

Type

entity type ContactStageJob extends BatchJob<ContactStageJob, BatchJobOptions, string> type key 'CSTGJ' {
  doStart:      ~ js-server
  processBatch: ~ js-server
  allComplete:  ~ js-server
}

Save the implementation as ContactStageJob.js, with one top-level function for each declared function. Replace buildContactCsv() with your own function that returns the CSV text, for example by joining or converting your source files:
JavaScript
```
function doStart(job, options) {
  job.scheduleBatch('contacts_staged.csv');
}

function processBatch(batch, job, options) {
  var fileName = batch && batch.value ? batch.value : batch;
  var inbox = FileSourceCollection.forName('contactsFileColl').inboxUrl();
  var csv = buildContactCsv();   // your join, conversion, or aggregation logic
  FileSystem.makeFile(inbox + fileName).writeString(csv);
}

function allComplete(job, options) {
  var inbox = FileSourceCollection.forName('contactsFileColl').inboxUrl();
  SourceFile.syncAll(inbox, DataIntegSpec.make({ process: true }));
  return 'staged file and started the pipeline';
}
```
processBatch writes the file to the inbox of the FileSourceCollection, and writeString overwrites any file with the same name. allComplete calls SourceFile.syncAll, which registers the new file and processes it through the transform. FileSystem.makeFile uses the application's default file system; if the collection's inbox is in Amazon S3, use FileSystem.s3().makeFile(...) instead.
The first line of processBatch unwraps the batch. When the batch Type is a primitive such as string or int, JavaScript can receive the batch boxed as { value: ... } instead of the bare value. A boxed value used as a string produces a path such as .../inbox/{"value":"..."}, and the write fails.
Sync the package, then start the job in the Console:
JavaScript
```
var job = ContactStageJob.make({}).upsert();
job.start(null);
```
Pass null as the options argument unless you defined your own options Type.
The job runs asynchronously. Read the job status with job.status(true) until status is completed or failed. When the job completes, syncAll has queued the staged file, and the pipeline processes it asynchronously. Confirm the load with the collection's SourceStatus records and a row count on Contact. For both checks, see Select a Pipeline Execution Type.

If a staging batch fails, allComplete does not run, so this job does not start the pipeline. The failed job stays in the failed status until you fix the cause and recover it. Files that successful batches wrote stay in the inbox, and any other sync of that inbox, such as a scheduled CronJob, loads them.

Run a job after the pipeline

The enrichment job runs after the pipeline loads rows: it reads the target entity Type and normalizes each phone number. The platform does not call a function when a pipeline run completes. Start the job after you confirm that the run finished, for example from the collection's SourceStatus records.

Save the Type as ContactEnrichJob.c3typ. Each batch is one Contact ID:

Type

/**
 * Post-pipeline enrichment job. Normalizes Contact phone numbers after a
 * data integration load. Each batch is one Contact ID.
 */
entity type ContactEnrichJob extends BatchJob<ContactEnrichJob, BatchJobOptions, string> type key 'CENRJ' {
  doStart:      ~ js-server
  processBatch: ~ js-server
  allComplete:  ~ js-server
}

Save the implementation as ContactEnrichJob.js:

JavaScript

function doStart(job, options) {
  Contact.fetch({ limit: -1 }).objs.each(function (c) {
    job.scheduleBatch(c.id);
  });
}

function processBatch(batch, job, options) {
  var id = batch && batch.value ? batch.value : batch;
  var c = Contact.forId(id).get('this');
  if (c && c.phone) {
    Contact.make({ id: c.id, phone: c.phone.split('-').join('') }).merge();
  }
}

function allComplete(job, options) {
  return 'checked phone numbers on ' + Contact.fetchCount() + ' contacts';
}

doStart schedules one batch for each Contact. fetch returns 2,000 rows unless you set a limit, so limit: -1 reads every row. For a large table, schedule batches from fetchObjStream instead, which streams the rows instead of loading them all into memory. merge() writes only the fields set on the object, so it updates phone and leaves the other fields unchanged.

Sync the package, then start the job in the Console:
JavaScript
```
var job = ContactEnrichJob.make({}).upsert();
job.start(null);
```

Read the job status, as for the staging job, until it reaches completed or failed:

JavaScript

var st = job.status(true);   // true = refresh with current progress
st.status;                   // "completed"
st.info;                     // "checked phone numbers on 10 contacts"

The info string is the return value of allComplete. Return a one-line summary there so the status shows what the job did.

You can chain more steps the same way: the allComplete of one job can start the next job, which builds a staged sequence without a scheduler.

Rebuild application dependencies after a historical load

After a large historical load, data derived from the loaded records can be missing or out of date: stored calculations, materialized hierarchies, and cached metrics. Data Fusion has a dialog that recomputes this derived data.

By default, every upsert queues its derived data for recomputation. During a backfill, the same calculated field can then be recomputed once per loaded record. To avoid the repeated recomputation, switch off async processing before the load:

In Data Fusion, select the settings button (gear icon) and turn on Disable Async Processing. The toggle is unavailable when the application runs in production mode.
In the Console, run InvalidationConfig.inst().setConfigValue('upsertAsyncProcessingDisabled', true, ConfigOverride.APP).

After the load finishes, turn the toggle off, or run the same Console call with false. Then rebuild the dependencies:

In Data Fusion, on the Data Integration tab, select the settings button (gear icon) next to Add Data Source.
Select Rebuild Application Dependencies. The Rebuild Application Dependencies dialog opens.
Under Configuration, select a preset:
- Post-Historical Load: the default, recommended after a backfill. The preset selects Stored Calculations, Hierarchy Denormalization, Timeseries Normalization, Metric Dependencies, Cached Metrics, and Analytics.
- Complete Rebuild: all nine dependency types.
- Calculations Only: Stored Calculations only.
- Custom: you select the dependency types yourself.
Review the dependency types. The table lists Stored Calculations, Hierarchy Denormalization, Timeseries Normalization, Metric Dependencies, Cached Metrics, Analytics, Created/Updated Callbacks, ACL Materialization, and KV Store Compaction. Keep at least one selected: with none selected, Start Rebuild is unavailable.
(Optional) Under Rebuild Behavior, select either option:
- Initial Load Mode: removes all existing metric dependency records (MetricDependency) before the rebuild. The removal can take significant time if many records exist.
- Fail on Data Issues: stops the rebuild on a calculated field error. By default, the rebuild records each error as a calculated field issue record (CalcFieldIssue) and continues.
(Optional) Expand Advanced to change the Batch Size and Sub-Batch Size for each stage. Keep the defaults unless a stage has performance issues. Higher values raise throughput but use more memory, so lower them if a stage runs out of memory or times out.
Select Start Rebuild, or Cancel to close the dialog without a rebuild.
The rebuild starts as a job, and the dialog shows App rebuild started successfully. Follow the job on the Batch Jobs page. Only one rebuild runs at a time. If a rebuild is already running, the dialog shows App Rebuild Already Running. Select View Current Status to follow the running rebuild, or Cancel Running Job to stop the running rebuild.

A rebuild needs at least two compute slots. When hierarchies are enabled, stored calculations rebuild in two phases.

You can also run the rebuild from the Console. App.rebuild({ storedCalcs: true }) starts a rebuild of stored calculations; see AppRebuildSpec for the other options. App.rebuildStatus() returns the rebuild progress, and App.cancelRebuild() stops the rebuild. Cancelling does not stop rebuild work for a Type that has already started.

Diagnose a failed job

To find why a job failed, read the errors on its status. When the status is failed, job.status(true) also returns an errors array with the error message for each failure:

JavaScript

var st = job.status(true);
st.status;               // "failed"
st.errors[0].errorMsg;   // the error message

The errors are also stored as InvalidationQueueError records. Fetch them by the job's ID to see which stage failed:

JavaScript

var errs = InvalidationQueueError.fetch({
  filter: Filter.eq('targetObjId', job.id)
}).objs;

errs[0].entryType;         // which stage failed, for example "dispatchDoStart" or "batch"
errs[0].detail.errorMsg;   // the error message

entryType names the failed stage. dispatchDoStart means the job never got past its start, for example because of a missing implementation binding or an error thrown in doStart. batch means one or more processBatch calls threw an error, and each has its own error record.

After you fix the cause, check that the status is failed, because recover() throws an error while the status is still failing. Then call job.recover() in the Console. Recovery runs only the failed batches again, then runs allComplete when every batch succeeds. If doStart failed, recovery runs the whole job again.

Fix common problems

Each row starts with what you see when a batch job fails:

Symptom	Cause	Fix
Every start fails; `entryType` is `dispatchDoStart` and the error says `doStart is not implemented`	Functions declared with a bare `~`, with no implementation language	Declare each function as `~ js-server` (for example `doStart: ~ js-server`) in the Type that extends `BatchJob`
`start(...)` fails with `Invalid call to start: ... Invalid value ... for parameter: options: BatchJobOptions`	A primitive, such as a number, passed as the options argument	Pass `null`, or define an options Type that mixes `BatchJobOptions`
Batch values arrive in an unexpected form, for example a file path containing `{"value":...}`	JavaScript can receive primitive batch values boxed	Unwrap with `batch && batch.value ? batch.value : batch`
`allComplete` never ran	At least one batch failed; `allComplete` runs only when every batch succeeds	Read the errors on the job status, fix the failing batch, then call `recover()`
A pipeline started from `processBatch` ran on partial data	Batches run in parallel, so the last batch to start is not always the last to finish	Start the pipeline from `allComplete`, not from `processBatch`

Copy link to this sectionPrerequisites

Copy link to this sectionUnderstand the batch job contract

Copy link to this sectionPrepare data and start the pipeline

Copy link to this sectionRun a job after the pipeline

Copy link to this sectionRebuild application dependencies after a historical load

Copy link to this sectionDiagnose a failed job

Copy link to this sectionFix common problems

Copy link to this sectionSee also

Prerequisites

Understand the batch job contract

Prepare data and start the pipeline

Run a job after the pipeline

Rebuild application dependencies after a historical load

Diagnose a failed job

Fix common problems

See also