Ingest Files from Amazon S3 Programmatically

Use this quickstart when your data lands as CSV files in an Amazon S3 bucket and you want to script the load. To build the same pipeline on the Data Fusion canvas, the visual pipeline editor in C3 AI Studio, see Connect with File Source System.

You connect the bucket, inspect a file, model the data, create a source collection, run the pipeline, and verify the stored entities. Later sections show how to make the pipeline part of your package and how to add a mount for your application from code. "Fix common problems" covers the errors you are most likely to see. The worked example loads an iris measurement CSV, but the steps apply to any delimited file. The example names (irisS3, irisS3Coll, IrisSource, and Iris) stand in for your own source system, collection, and Type names.

You work in two places. You run runtime commands in the C3 AI Console of your application. You define Types and the transform in your IDE as package source, which the C3 Agentic AI Platform syncs to the application.

A file-source pipeline links these Types:

• FileSourceSystem: the source system, which is the connection to the S3 bucket.
• FileSourceCollection: the source collection, which is a folder of files that share one schema.
• Source: the Source Type, which describes one row of a source file.
• Transform: the transform, which maps each source row to the target.
• A target entity Type: the Type that stores the data in the application.

Data passes through these five Types in the order listed. The source system reaches the bucket, and the source collection picks a folder and its Source Type. The transform writes each row to the target entity Type.

Copy link to this section

Prerequisites

Before you start, make sure you have the following:

• A C3 Agentic AI Platform application in dev mode, with access to the C3 AI Console. The platform accepts new Types, transforms, source systems, and source collections only while the application is in dev mode.
• A role that allows data integration from code. To choose one, see Data Integration Roles and Permissions.
• Amazon S3 credentials, an access key and a secret key, with read access to the bucket.
• CSV files in a folder of the bucket. The example uses the iris dataset, uploaded as iris_1.csv and further files.
• The package source for your application, open in your IDE, such as VS Code with the C3 AI extension.

Copy link to this section

Step 1: Connect to Amazon S3

In the C3 AI Console, declare the bucket as a source system and set its credentials. The source system's rootUrlOverride tells the platform which bucket the source system reads. Every later step reads the bucket through this connection.

var fss = FileSourceSystem.make({
  name: 'irisS3',
  rootUrlOverride: 's3://<your-bucket>/'
}).upsert();

fss.setCredentials(
  AwsCredentials.fromAccessKey('<access-key>', '<secret-key>', '<region>'),
  false
);

FileSystem.s3().listFiles('s3://<your-bucket>/');

Replace the following placeholders with your values:

• <your-bucket>: The S3 bucket name.
• <access-key> and <secret-key>: The Amazon S3 credentials.
• <region>: The bucket region, for example us-west-2.

Each call does one job:

• upsert saves the source system with its bucket URL.
• setCredentials stores the keys for the bucket in your application. The second argument, false, replaces any credentials stored earlier instead of merging with them. You do not need cluster administrator access for this call. Other routes, such as FileSystem.s3().enable() and AwsS3Bucket.setCredentialsForResourceName, require cluster administrator access.
• listFiles tests the connection. The platform stores credentials without validating them, so a wrong key fails only on first use. A working connection returns an object with a files array, one entry per file or folder.

The platform stores credentials per bucket, not per source system. A second source system on the same bucket reuses the stored credentials, and a different bucket needs its own setCredentials call.

Copy link to this section

Step 2: Inspect the source data

In the Console, read the first 400 characters of one file. The Source Type you write in the next step must match the column headers exactly, because the platform ignores a column that matches no field.

var sampleUrl = 's3://<your-bucket>/<path-to-files>/iris_1.csv';
FileSystem.s3().makeFile(sampleUrl).readString(0, 400);

The call returns the header row and the first rows of the file as a string.

Take the <path-to-files> value from the listFiles output in Step 1. Each entry in the files array has a url. Use the folder portion of that URL, between the bucket name and the file name. Copy the path instead of typing it, because listFiles returns an empty result for a mistyped folder instead of an error.

Address the file through the S3 file system, as shown. FileSystem.makeFile uses the default application file system and rejects an s3:// URL.

The example file has seven columns:

• Id: an integer row identifier.
• SepalLengthCm, SepalWidthCm, PetalLengthCm, PetalWidthCm: the measurements.
• Species: the species name, for example Iris-setosa.
• Encoded: an integer species code.

Copy link to this section

Step 3: Model the data

Define what a source row looks like, what the application stores, and how one maps to the other. In your IDE, add three files to the package source, then sync the package to the application.

Define the Source Type in src/IrisSource.c3typ. The Source Type mirrors the file as the file arrives: one field per column, with the column's data, not the shape you want to store. The column headers are PascalCase, so each field uses the @ser annotation to name the column it reads:

type IrisSource mixes Source {
  @ser(name="Id") irisId: string
  @ser(name="SepalLengthCm") sepalLengthCm: double
  @ser(name="SepalWidthCm") sepalWidthCm: double
  @ser(name="PetalLengthCm") petalLengthCm: double
  @ser(name="PetalWidthCm") petalWidthCm: double
  @ser(name="Species") species: string
  @ser(name="Encoded") encoded: int
}

Define the target entity Type in src/Iris.c3typ. The application stores and queries this Type, and the pipeline fills it. The schema name clause is optional; it overrides the default database table name.

entity type Iris schema name 'IRIS' {
  sepalLengthCm: double
  sepalWidthCm: double
  petalLengthCm: double
  petalWidthCm: double
  species: string
  encoded: int
}

Define the transform that maps source rows to the target. Save the transform as metadata/Transform/IrisSource-Iris.js:

data = {
  name: 'IrisSource-Iris',
  source: 'IrisSource',
  target: 'Iris',
  projection: {
    id: concat('iris_', irisId),
    sepalLengthCm: sepalLengthCm,
    sepalWidthCm: sepalWidthCm,
    petalLengthCm: petalLengthCm,
    petalWidthCm: petalWidthCm,
    species: species,
    encoded: encoded,
  },
};

The projection lists each target field and the expression that fills it from the source row. The id expression decides what happens when a row loads again. The platform upserts each row: a row whose id already exists updates that entity instead of creating a duplicate. An id built from the row's own values therefore makes reprocessing safe. To keep every row separate instead, add a file-unique component to the id.

Save the three files and wait for the platform to finish syncing the package source. Once the platform has deployed the Type, Iris.fetchCount() runs in the Console and returns 0 before the first load.

Copy link to this section

Step 4: Create the source collection

In the Console, create the source collection. The source collection links the connection from Step 1 to the Source Type from Step 3: the platform reads each file in the collection's folder as IrisSource rows. Then point the source collection at the folder that holds your files and confirm the result.

FileSourceCollection.make({
  name: 'irisS3Coll',
  source: IrisSource,
  sourceSystem: FileSourceSystem.forName('irisS3')
}).upsert();

FileSourceCollection.forName('irisS3Coll').config().setConfigValue(
  'inboxUrlOverride',
  's3://<your-bucket>/<path-to-files>/',
  ConfigOverride.APP
);

FileSourceCollection.forName('irisS3Coll').inboxUrl();

The inbox is the folder the source collection reads files from. By default, the inbox is the source system root URL followed by the collection name and inbox. Set inboxUrlOverride only when your files sit elsewhere. Use the same <path-to-files> value you copied from the listFiles output in Step 1. inboxUrl returns s3://<your-bucket>/<path-to-files>/ when the override is in place.

The Console route suits iteration in a dev application. To deploy the pipeline with your package instead, follow "Make the pipeline part of your package".

Copy link to this section

Step 5: Run the pipeline

In the Console, sync the inbox and process the files to load the data into Iris. Syncing creates a SourceFile record for each file in the inbox. Processing reads the rows, applies the transform, and writes Iris entities.

First, confirm that the inbox holds files:

var inbox = FileSourceCollection.forName('irisS3Coll').inboxUrl();
FileSystem.countFiles(inbox);

If countFiles returns 0, fix inboxUrlOverride before you continue. A sync against an empty or wrong path registers no files and reports no error.

Then sync and process the files:

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

The process flag on the DataIntegSpec runs the pipeline right after the sync. syncAll runs in the background. syncAll returns a reference to the batch job that does the work, and the rows arrive in Iris when the batch job finishes.

To check the registered files before you load them, omit the process flag. Synced files then stay in the initial state, which means synced but not processed. Run SourceFile.processAll() to process them, or pass a subset of the SourceFile records to SourceFile.processBatch.

Copy link to this section

Step 6: Verify the data

Check the status records to find any stage that failed, then fetch the entities to confirm that the data loaded.

SourceStatus.fetch({
  filter: "sourceCollectionName=='irisS3Coll' && typeIdent=='SOURCE'",
  order: 'descending(meta.updated)'
});

Each SourceStatus record reports state, count, successful, and failed for one source file. count, successful, and failed are row counts. A completed state with failed at 0 indicates a clean run. For the example, each file reports completed with failed at 0.

Read the persisted entities:

c3Grid(Iris.fetch({ include: 'id, species, sepalLengthCm, petalWidthCm', limit: 100 }));

The grid shows the Iris records with their measurements and species. Your application can now query Iris like any other entity Type.

To make the run fail on a column that matches no Source Type field, set failOnExtraFields on the source collection's SourceCollection.Config:

FileSourceCollection.forName('irisS3Coll').config().setConfigValue('failOnExtraFields', true, ConfigOverride.APP);

failOnExtraFields catches extra or misspelled columns. A Source Type field that has no column in the file stays empty without an error.

Copy link to this section

Make the pipeline part of your package

Declare the source system and source collection as package metadata when the pipeline must deploy with your package. The Console route in Steps 1 and 4 suits iteration in a dev application. The next package deploy replaces a source system or source collection that you created in the Console.

The folder under metadata/ or config/ must match the Type name, and each file carries the same fields you passed to make:

1. Create metadata/FileSourceSystem/irisS3.json:

   JSON

   Copy code to clipboard

   { "name": "irisS3", "rootUrlOverride": "s3://<your-bucket>/" }

2. Create metadata/FileSourceCollection/irisS3Coll.json:

   JSON

   Copy code to clipboard

   {
     "name": "irisS3Coll",
     "source": "IrisSource",
     "sourceSystem": { "type": "FileSourceSystem", "name": "irisS3" }
   }

3. If your files sit outside the default inbox, create config/SourceCollection.Config/irisS3Coll.json:

   JSON

   Copy code to clipboard

   { "name": "irisS3Coll", "inboxUrlOverride": "s3://<your-bucket>/<path-to-files>/" }

4. Sync the package to the application.

Keep credentials out of the package. Credentials are secrets: set them at runtime with setCredentials, as in Step 1.

Copy link to this section

Add a mount at the application level

A mount is a named shortcut to a storage URL. To add your own mount for the application from the Console, call setMount with ConfigOverride.APP, then read it back with mounts(). This example mounts a folder inside the application's default storage:

var fsys = FileSystem.inst();
var target = fsys.mounts().get('/') + 'orders_landing/';
fsys.setMount('ordersLanding', target, ConfigOverride.APP);

fsys.mounts().get('ordersLanding');      // returns the target URL
fsys.mounts().keys().collect();          // lists every mount name

mounts().keys() returns a stream, so call .collect() before you use array methods on it. To read and write through the mount, build file URLs from mounts().get('ordersLanding').

Each file system has its own accessor, such as FileSystem.s3(), FileSystem.gcs(), and FileSystem.azure(). FileSystem.inst() returns the default file system of the environment. To turn on a file system that is not enabled, call enable() on its accessor. The call acts on the whole environment, so it needs cluster administrator access.

Copy link to this section

Fix common problems

Find the symptom you see, then apply the fix in the same row.

Copy link to this section

See also

• Integrate Data Programmatically
• Connect with File Source System
• Monitor Your Data Pipeline
• Select a Pipeline Execution Type
• Ingest a Kafka Message Stream Programmatically
• Combine Multiple Sources with a Canonical