# Ingest data

Ingest turns a file into rows in a <Term id="nexus">Nexus</Term> table. Nexus reads the file, works out columns and types, cleans the values and writes an <Term id="iceberg">Apache Iceberg</Term> table. You can preview first, choose how rows are added and follow the run.

## Before you start [#before-you-start]

- Your key's **Access** allows writing to the target <Term id="schema">schema</Term>.
- The workspace has <Term id="credits">credits</Term> and room to store the data.
- The file is one Nexus reads, such as CSV, TSV, Parquet, JSON, JSON Lines, Excel, GeoJSON, GeoPackage or a shapefile.

## Load a file into a table [#load-a-file-into-a-table]

<Steps>
<Step>
### Preview it [#preview-it]

`ingest_preview()` reads a sample and reports what it found, writing nothing:

```python
from klarety import Nexus

with Nexus.connect() as nexus:
    preview = nexus.ingest_preview("sales_2026.csv", into="sales")
    print(preview["source_format"], preview["rows_sampled"])
    for column in preview["columns"]:
        print(column)
```

Check the column names and types. If one is wrong, fix the file first.
</Step>
<Step>
### Start the load [#start-the-load]

```python
run = nexus.ingest("sales_2026.csv", into="sales", key=["order_id"])
print(run.run_id, run.state)
```

A local path uploads to storage in parts, so files larger than memory are fine. `ingest()` returns an `IngestRun` as soon as the load is accepted.
</Step>
<Step>
### Follow it to the end [#follow-it-to-the-end]

```python
while run.state in ("accepted", "running"):
    time.sleep(5)
    run = nexus.ingest_status(run.run_id)
print("Finished:", run.state)
```

A run ends as `succeeded`, `failed` or `cancelled`.
</Step>
<Step>
### Check the rows [#check-the-rows]

```python
print(nexus.sql("SELECT COUNT(*) AS n FROM sales").rows)
```

The script below puts the four steps together.
</Step>
</Steps>

### The whole script [#the-whole-script]

```python title="load_sales.py"
import sys
import time

from klarety import InsufficientCreditsError, Nexus, RefusedError, ServiceError

PATH, TABLE = "sales_2026.csv", "sales"

with Nexus.connect() as nexus:
    try:
        run = nexus.ingest(PATH, into=TABLE, key=["order_id"])
        while run.state in ("accepted", "running"):
            time.sleep(5)
            run = nexus.ingest_status(run.run_id)
    except FileNotFoundError:
        sys.exit(f"{PATH} is neither a file nor a uri")
    except InsufficientCreditsError:
        sys.exit("The workspace is out of credits")
    except RefusedError as refused:
        sys.exit(f"Load refused: {refused}")
    except ServiceError as failed:
        sys.exit(f"Load failed on our side, try again: {failed}")

    if run.state != "succeeded":
        sys.exit(f"Load ended as {run.state}")
    print(nexus.sql(f"SELECT COUNT(*) AS n FROM {TABLE}").rows[0]["n"], "rows")
```

## Choose how rows are added [#choose-how-rows-are-added]

`mode` decides what happens when the table already has rows:

| Mode | What it does |
| --- | --- |
| `auto` (default) | A first load creates the table. After that, with a `key` it updates rows whose key exists and adds the rest; without a key it adds only rows that are not already there. |
| `append` | Adds every row, even if the same row is already there. |
| `upsert` | Updates rows whose `key` exists and adds the rest. Give a `key`. |
| `replace` | Replaces every row in the table with the file's rows. |

`key` names the columns that identify a row, such as `["order_id"]`. Other parameters:

- `into`: the table to load into.
- `schema`: the schema of the table, `default` unless you name another.
- `force`: reload a file already loaded into this table. Without it, a file loads into a table only once, so rerunning your script is harmless.

## Load from other places [#load-from-other-places]

```python
# A file already in a folder in Object Storage: every file directly in it
run = nexus.folders.at("exports/daily").load(into="daily_exports", mode="append")

# Chosen files by their IDs
run = nexus.files.load(["file-id-1", "file-id-2"], into="daily_exports")

# A file in an Amazon S3 bucket you connected
result = nexus.connections.read(connection_id, "exports/2026-11.parquet", into="exports")
```

To load each new file as it lands in a folder, use `folder.watch(into="table")`. See [Files and folders](/docs/developers/files) and [Connections](/docs/developers/connections).

## Load a Spark DataFrame [#load-a-spark-dataframe]

If you prepared data in a [Spark session](/docs/developers/spark), load the DataFrame directly:

```python
run = nexus.ingest_frame(frame, table="customer_summary", key=["customer"])
```

## What can go wrong [#what-can-go-wrong]

| Problem | Cause | Fix |
| --- | --- | --- |
| `FileNotFoundError` | The path is not a file and not a uri. | Check the path. |
| `RefusedError` | No write access, or the file is larger than your plan allows for one upload. The message says which. | Use a key with access, or connect the bucket the file lives in and read it from there. |
| Run ends `failed` | The file could not be read. | Preview it, fix it, load again. |
| Nothing new loaded | The same file was already loaded into this table. | Pass `force=True` if you mean to load it again. |

## Next steps [#next-steps]

- [Tables and schemas](/docs/developers/tables) to set rules and describe columns.
- [Files and folders](/docs/developers/files) to store files and watch folders.
- [CLI](/docs/developers/cli) to load a file from your terminal with `klarety ingest`.