Monitor Your Data Pipeline

After you run a Structured Data Integration (SDI) pipeline, check that its data arrived. The Data Monitoring page in C3 AI Studio shows data volume and failed loads. You check the volume for the last seven days and find a source collection in the Source Collections grid. Then you open the collection to see the status of every file it loaded. If you build pipelines in code, you can run the same checks from the C3 AI Console. The File states section lists every state a file can show. The Data Monitoring page covers structured pipelines only. For Unstructured Data Integration (UDI) pipelines, see the Monitor an Unstructured Data Integration pipeline section.

A source collection is the set of files or messages that a pipeline reads from one source system. Each file or message in a source collection is a source object. A row is one line or entry read from a source object. The pipeline's transform maps the rows to records, and a record is an instance of the target entity Type that the pipeline creates or updates.

Copy link to this section

Prerequisites

Before you open the page, you need:

• An SDI pipeline that has run at least once. To build and run one, see Configure File Sources. For how a run works, see Data Integration End-to-End Workflow. Until a pipeline runs, the chart shows "No volume data available".
• Access to the application in C3 AI Studio with a developer or administrator role. The Data Monitoring item appears only for these roles. For the roles, see Data Integration Roles and Permissions.

Copy link to this section

Open the Data Monitoring page

Open the page from your application in C3 AI Studio:

1. In C3 AI Studio, open your application.
2. In the navigation pane, select Data > Data Monitoring.

The Data Monitoring page opens with the Data volume (last 7 days) chart and the Source Collections grid.

Copy link to this section

Check data volume

Use the Data volume (last 7 days) chart to see how much data your source collections loaded each day. The chart covers all source collections in the application and plots one line for each series:

A rise in Failed rows or Load errors tells you to check the affected source collection in the grid.

Copy link to this section

Review source collections

The Source Collections grid lists file, cloud message, and API source collections with their state and recent throughput. SQL source collections do not appear in the grid. Check SQL source collections from the Console, as described in Check a run from the Console. Narrow the grid with these controls:

• Search: enter part of a collection name.
• Category: select File or Cloud Message to show only collections of that category.
• Refresh: select the refresh button to load the latest data. The page loads data only when you open it or select refresh.

Each row in the grid shows one source collection:

A failure rate that rises across refreshes shows a collection that has started to fail.

Copy link to this section

Investigate failed files

Open a source collection to find the files that did not load and the errors they recorded. For a file source collection:

1. On the Data Monitoring page, select the collection name in the Source Collections grid.

   The collection's page opens. Check the collection's state, for example OK.

2. In Source Collection Status, check the Failed and Rejected counts.

   The tiles count the collection's files: Total Files, Completed, Processing, Failed, and Rejected. A count greater than zero in Failed or Rejected means that some files did not load.

3. In the Files grid, select Failed to list only the files that failed. Rejected files do not appear under Failed. To find rejected files, clear the filter and check the Status column.

   Each row shows File Name, Status, Started, Completed, Error, Total, Succeeded, and Failed. To find one file, enter its name in Search by file name.

4. Select a file name. A panel opens with the file's counts and two tabs, Status Rollup and Errors. Select Errors to read the errors that the file recorded. To open the details as a full page, select View Details.

5. Fix the cause of the error:

   • Correct the transform in Data Fusion where you built the pipeline.
   • Fix the source data.
   • Update the source system credentials. For credentials, see Authenticate File Sources, Authenticate Database Sources, or Authenticate Streaming Sources.

6. Run the failed files again. A new run processes only new or changed files, so it does not pick up a failed file. By default the platform also moves each processed file, including a failed one, out of the inbox to the archive. Retry the failed files in one of two ways:

   • On the Data Fusion canvas, select Execute pipeline on the source collection node. In the Configure Runtime Parameters dialog, choose Select files, filter Processing Status to Failed, select the failed files, and select Execute.

   • In the Console, process the collection's failed files. Replace ordersFileColl with the name of your source collection:

     JavaScript

     Copy code to clipboard

     SourceFile.processAll(DataIntegSpec.make({
       filter: "sourceCollectionName=='ordersFileColl' && status.state=='failed'"
     }));

     To retry rejected files, use status.state=='rejected' in the filter instead.

   If a file still fails, or never leaves Initial, find its symptom in the Fix common problems table of Configure a Source Collection.

7. Return to the collection's page, select the refresh button in Files, and check the Status of the files from the new run.

The Data Monitoring page shows load results and does not reprocess data. To load data again, run the pipeline.

Copy link to this section

File states

The Status column in the Files grid shows the state of each file (source object). A state starts at Initial and ends at Completed, Failed, Rejected, or Stopped:

Copy link to this section

Check a run from the Console

If you build pipelines in code, you can check the same results from the Console. Replace ordersFileColl with the name of your source collection, then run these checks:

1. Check the processing queues. SourceQueue.countAll() returns the counts of the source queue, including pending, failed, and unreviewedFailed. InvalidationQueue.countAll() returns one entry for each queue, such as SourceQueue, BatchQueue, and MapReduceQueue.

2. List the latest pipeline statuses. Each DataIntegStatus record names one stage of a run, such as SOURCE, SOURCE:CHUNK, TRANSFORM, or TARGET, with its state:

   JavaScript

   Copy code to clipboard

   c3Grid(DataIntegStatus.fetch({ order: 'descending(meta.created)', limit: 20 }));

3. Find the failed SourceStatus records of your collection:

   JavaScript

   Copy code to clipboard

   c3Grid(SourceStatus.fetch({
     filter: "state=='failed' && sourceCollectionName=='ordersFileColl'"
   }));

4. Read the record errors on TargetStatus. A record that fails to load into the target entity Type reports its error in TargetStatus.errors, with the field errors and the source row. The source status record of the same file shows the failed state, and its firstError can be empty. Read the latest failed target status:

   JavaScript

   Copy code to clipboard

   var ts = TargetStatus.fetch({ filter: "state=='failed'", order: 'descending(meta.created)', limit: 1 }).objs[0];
   ts.errors[0].error.message;   // for example "Required field Order.region not set."

5. To see every stage of one run, call the rollups on the file's root status record. Replace <status ID> with the ID of a step 3 result whose typeIdent is SOURCE. The step 3 results can also list chunk records, whose typeIdent is SOURCE:CHUNK, and the rollups fail with an error on a chunk record ID. SourceStatus.make({ id: '<status ID>' }).all() returns the status, and allChunks(), allTransforms(), and allTargets() return the chunk, transform, and target statuses.

If a file does not load at all, check for these three causes:

• The file has no SourceFile record: the platform has not synced it yet. Run SourceFile.syncAll(null) to sync the files of every file source collection.

• The file stays at initial: it was synced without being processed. Process the collection's waiting files:

  JavaScript

  Copy code to clipboard

  SourceFile.processAll(DataIntegSpec.make({
    filter: "sourceCollectionName=='ordersFileColl' && status.state=='initial'"
  }));

• The file stays at processing: check the SourceQueue entry that InvalidationQueue.countAll() returns. If the entry's paused value is true, the queue holds the work. If nobody paused the queue on purpose, run SourceQueue.resume().

The errors field of a status record holds only the first 100 errors. To read every error of a file, call allErrors() on its root status record, for example SourceStatus.make({ id: '<status ID>' }).allErrors().

A file source collection also keeps the source objects that failed, together with their errors. To list them, call failedSources() on the collection:

c3Grid(FileSourceCollection.forName('ordersFileColl').failedSources().collect());

Each result names the failed source, its error, and the target Type or transform where the error occurred. The platform keeps these records unless the collection's doNotArchiveFailedSources option is set. To delete them, call deleteAllFailedSources(true) on the collection.

Status records accumulate with each run, and the platform does not delete them. Count the source status records of one collection with fetchCount:

SourceStatus.fetchCount({ filter: "sourceCollectionName=='ordersFileColl'" });

To remove status records, call DataIntegStatus.removeAll({}, true). This call removes the status records of every source collection in the application, not only one. Reset Run State in Data Fusion runs this call and also clears the source queue with SourceQueue.clear(), which drops pending work for every collection. Remove the records only when you no longer need the history of earlier runs.

To remove the status records of one collection only, filter on the status IDs, which start with the collection name:

DataIntegStatus.removeAll({
  filter: "startsWith(id, 'ordersFileColl_') || startsWith(rootSourceStatus.id, 'ordersFileColl_')"
}, true);

The filter also matches any collection whose name starts with the same text, such as ordersFileColl_eu. DataIntegStatus has no sourceCollectionName field, so a filter on that field fails.

Copy link to this section

Get notified about failures

Data Fusion does not send a notification when a source collection fails to load. Check the Data Monitoring page, or run the Console checks above, after each scheduled run. To be notified about problems in the data that a pipeline loads, deploy a data validation rule with Send notifications to selected recipients turned on, as described in Validate Your Data Pipeline.

Copy link to this section

Monitor an Unstructured Data Integration pipeline

The Data Monitoring page lists structured source collections only. Unstructured Data Integration (UDI) collections don't appear in the grid. To monitor a UDI pipeline, use the Data Fusion canvas or select Monitoring > Jobs > Batch. For details, see Monitor Pipeline Execution.

Copy link to this section

See also

• Structured Data Integration Overview
• Validate Your Data Pipeline