> ## Documentation Index > Fetch the complete documentation index at: https://docs.nekt.com/llms.txt > Use this file to discover all available pages before exploring further. # Cloud Storage Files as a data source > Bring data files stored in Google Cloud Storage to Nekt, including Avro files written by Pub/Sub Cloud Storage subscriptions. Cloud Storage Files reads data files from a Google Cloud Storage bucket into tables in your Catalog. Each table you configure reads a set of files from the bucket, selected by folder and file name pattern. The first supported format is **Avro**, including the files a [Pub/Sub Cloud Storage subscription](https://cloud.google.com/pubsub/docs/cloudstorage) writes to a bucket. A common use is change data capture: database change events published to Pub/Sub, written to a bucket, and loaded into Nekt as one table of events. | What | Supported | | :- | :- | | Storage | Google Cloud Storage | | File formats | Avro (any codec), also when gzip-compressed | | Envelopes | None, or Pub/Sub Cloud Storage subscription | | Sync types | INCREMENTAL, based on when each file was written to the bucket | ## Prerequisites You need a Google Cloud service account that can read the bucket, and a JSON key for it. **1. Create a service account** In the Google Cloud console, go to **IAM & Admin > Service Accounts** in the project of your choice and create a service account, for example `nekt-reader`. **2. Grant read access on the bucket** Open the bucket in **Cloud Storage > Buckets**, go to **Permissions**, click **Grant access**, and give the service account the **Storage Object Viewer** role (`roles/storage.objectViewer`). Granting it on the bucket, rather than on the project, limits Nekt to that bucket. **3. Create a JSON key** In the service account, go to **Keys > Add key > Create new key**, choose **JSON**, and download the file. You will upload it in the next section. Google's guide: [Create and delete service account keys](https://cloud.google.com/iam/docs/keys-create-delete). > [!NOTE] > > Nekt only lists and downloads files. The connector requests a read-only Cloud Storage scope even if the service account has broader roles. ## Configuring Cloud Storage Files as a Source In the [Sources](https://app.nekt.ai/sources) tab, click on the "Add source" button located on the top right of your screen. Then, select the Cloud Storage Files option from the list of connectors. Click **Next** and you'll be prompted to add your access. ### 1. Add your access * **Storage**: `Google Cloud Storage`. * **Bucket**: the bucket name, with or without the `gs://` prefix. * **Google Cloud service account key (JSON)**: upload the JSON key file from the prerequisites. #### Tables Click **Add a table** once for each table you want. Each table reads a set of files from the bucket. * **Table name**: name of the table the files load into. Lowercase letters, digits, and underscores, starting with a letter, and unique within the source. * **Folder**: only files whose name starts with this prefix are read, e.g. `exports/cdc/`. Leave empty to read from the root of the bucket. * **File pattern**: (Default: `.*`) a regular expression searched in each file name, folder included. Only matching files are read. `.*` reads every file under the folder; `\.avro$` reads only files ending in `.avro`. * **File format**: `Avro`. * **Envelope**: (Default: `None`) choose **Pub/Sub Cloud Storage subscription** when the files were written by a Pub/Sub Cloud Storage subscription. Each record in those files holds one message, and the message body becomes the payload. Choose **None** when each record in the file is the data itself. * **Message body format**: (Default: `JSON`) with the Pub/Sub envelope, **JSON** parses each message body, so its fields can be promoted to columns. **Text** keeps the body as plain text. * **Schema mode**: (Default: `Raw`) each record is stored whole, as JSON text, in the `payload` column, plus the promoted columns. Use it when records do not share one schema, for example change events of several collections in the same files. * **Promoted columns**: values copied out of each record into their own columns, so the table can be filtered without parsing JSON. **Key** is the column name (lowercase snake\_case) and **Value** is the [JSONPath](https://goessner.net/articles/JsonPath/) of the value in the record, e.g. `$.collection` or `$.documentKey._id`. Promoted columns are always stored as text; objects and arrays are stored as JSON text. * **Primary keys**: (Default: `_nekt_file_path`, `_nekt_file_row`) columns that identify a record. The default makes re-reading a file update its rows instead of duplicating them. Use `_nekt_record_hash` to also collapse identical records delivered in different files, as Pub/Sub does. * **Skip if empty**: (Default: off) when no file matches the folder and pattern, skip this table with a warning instead of failing the run. > **Example: MongoDB change events from a Pub/Sub subscription** > > Change events published to Pub/Sub and written by a Cloud Storage subscription under `cdc/`, one row per event: > > | Field | Value | > | :- | :- | > | Table name | `mongo_change_events` | > | Folder | `cdc/` | > | File format | `Avro` | > | Envelope | `Pub/Sub Cloud Storage subscription` | > | Promoted columns | `operation_type` → `$.operationType`, `database` → `$.database`, `collection` → `$.collection`, `document_id` → `$.documentKey._id`, `cluster_time` → `$.clusterTime` | > | Primary keys | `_nekt_record_hash` | > > Update events usually carry only the fields that changed. To rebuild the current state of each collection, use a [Notebook](https://docs.nekt.com/notebooks/overview.md) or a [Query](https://docs.nekt.com/queries/overview.md) that applies the events in `cluster_time` order. #### Advanced settings * **Start date**: only files written to the bucket at or after this moment are read. Leave empty to read every file. * **Maximum files per run**: (Default: `10000`) the most new files one run reads per table, oldest first. A long history is loaded over several runs, each small enough to finish. * **Parallel downloads**: (Default: `8`, up to `64`) how many files are downloaded at the same time. Once you're done, click **Next**. Nekt checks that the service account can list the bucket before moving on. ### 2. Select streams You will see one stream per table you configured. Select the ones you want and click **Next**. ### 3. Configure data streams Customize how you want your data to appear in your catalog. Select the desired layer where the data will be placed, a folder to organize it inside the layer, a name for each table (which will effectively contain the fetched data) and the type of sync. * **Layer**: choose between the existing layers on your catalog. This is where you will find your new extracted tables as the extraction runs successfully. * **Folder**: a folder can be created inside the selected layer to group all tables being created from this new data source. * **Table name**: we suggest a name, but feel free to customize it. You have the option to add a **prefix** to all tables at once and make this process faster! * **Sync Type**: INCREMENTAL, based on when each file was written to the bucket. Read more about Sync Types [here](https://docs.nekt.com/runs/sync-types.md). Once you are done configuring, click **Next**. ### 4. Configure data source Describe your data source for easy identification within your organization, not exceeding 140 characters. To define your [Trigger](https://docs.nekt.com/runs/scheduling-and-triggers.md), consider how often you want data to be extracted from this source. This decision usually depends on how frequently new files land in the bucket. Once you are ready, click **Next** to finalize the setup. ### 5. Check your new source You can view your new source on the [Sources](https://app.nekt.ai/sources) page. If needed, manually trigger the source extraction by clicking on the arrow button. Once executed, your data will appear in your Catalog. > [!WARNING] > > For you to be able to see it on your [Catalog](https://app.nekt.ai/catalog), you need at least one successful source run. ## Incremental sync * Nekt tracks the time each file was written to the bucket, not the content of its records. A file that arrives late is still read, even if its records are old. * Each run reads new files oldest first, up to **Maximum files per run** per table. When there are more, the run log says how many are left, and the next runs continue from where it stopped. A bucket with a long history is loaded over several runs. * The newest files already read are read again on the next run. Their rows are updated by primary key, not duplicated. * If a run fails, the next one reads the same files again. No file is skipped. > [!TIP] > > For a large history, schedule the source to run frequently until it catches up, then move it to the cadence you need. # Streams and Fields Each table you configure becomes one stream, named after its **Table name**. > **Columns of every table** > > **File metadata:** > > * `_nekt_file_path` - Location of the file the record came from, as `gs://bucket/path` > * `_nekt_file_last_modified` - When the file was written to the bucket; used as the incremental replication key > * `_nekt_file_row` - Position of the record in its file, starting at 1 > > **Record:** > > * `_nekt_record_hash` - SHA-256 of the record's original bytes (with the Pub/Sub envelope, of the message body). Identical records have the same hash, even in different files > * `payload` - The record as JSON text. With the Pub/Sub envelope, the message body exactly as it was published > > **Pub/Sub message metadata** (only with the Pub/Sub envelope, and filled only when the subscription has **Write metadata** enabled): > > * `pubsub_message_id` - ID Pub/Sub assigned to the message > * `pubsub_publish_time` - When the message was published > * `pubsub_attributes` - Message attributes, as JSON text > > **Promoted columns:** > > * One text column per entry in **Promoted columns**, named after its **Key** ## Troubleshooting | Error or symptom | What it means | What to do | | :- | :- | :- | | "The service account cannot read the bucket" | The service account has no read access to the bucket. | Grant it the **Storage Object Viewer** role on the bucket. | | "The bucket does not exist, or the service account's project cannot see it" | The bucket name is wrong, or it belongs to an organization the service account cannot reach. | Check the **Bucket** field. | | "Google rejected the service account key" | The key was deleted or rotated in Google Cloud. | Create a new JSON key and upload it in the source settings. | | The run says there are more new files than one run reads | The bucket has more new files than **Maximum files per run**. | Nothing to fix: the next runs load the rest. Raise the limit or run more often to catch up faster. | | The same event appears in more than one row | Pub/Sub delivers at least once, so a message can land in two files. | Use `_nekt_record_hash` as the primary key. | ## Skills for agents > **[Download Cloud Storage Files skills file](https://docs.nekt.com/sources/files.md)** > > Cloud Storage Files connector documentation as plain markdown, for use in AI agent contexts. This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.