C3 AI Documentation Home

Create and Load Data Lakehouse Tables

A Data Lakehouse table is an Iceberg table stored in a catalog. The fastest way to create a table is the Create new table wizard in C3 AI Studio. For richer options — primary keys, partition expressions, sort orders, merge writes, loading from sources other than a CSV file — switch to the notebook path documented in From a notebook.

In C3 AI Studio

Use the Create new table wizard when you have a CSV file and want to land it in the default catalog with minimal configuration.

  1. Open your application in C3 AI Studio.
  2. Select Data Lakehouse in the Data section of the side navigation.
  3. On the Tables tab, select Owned by me.
  4. Select Create new table.
  5. Select Browse, and choose one or more CSV files. The wizard supports multiple files for a single table; the schemas must match.
  6. Verify the parsed information:
    • Files: Every file in the upload is listed.
    • Schema: Column names and inferred data types.
    • Target preview: The first rows as they will land in the table.
    • File settings: Override the delimiter if the preview is wrong.
  7. Select Next.
  8. Enter a Table name and Description.
  9. Select Upload.

The wizard creates the table in the default catalog and loads the CSV in one step. The new table appears on the Tables tab and is queryable from the SQL Editor tab.

The Studio wizard supports the most common case: append a CSV into the default catalog. For primary keys, partition expressions, sort orders, merge writes, or loading from a source other than a CSV upload, use the notebook path below.

From a notebook

The notebook path exposes the full DataLake.CreateTableSpec surface and every supported load mode. Use it when you need primary keys, custom partitions, sort orders, or a data source other than a CSV upload.

Create a table from a schema

Define the column shape with ValueType, build a DataLake.CreateTableSpec, and create the table on the catalog.

Python
catalog = c3.DataLake.Catalog.inst()

schema = c3.ValueType.fromString(
    "{t: datetime, activePower: double, rotorSpeed: double, "
    "gearOilTemp: double, asset: string}"
)

spec = (
    c3.DataLake.CreateTableSpec.builder()
    .name("turbineMeasurements")
    .schema(schema)
    .build()
)

table = catalog.createTable(spec)

Create a table from a Data Fusion Source

To inherit the schema from an existing Source Type instead of declaring it by hand:

Python
spec = (
    c3.DataLake.CreateTableSpec
    .fromSource("CanonicalWindTurbineMeasurement")
    .withPartitionExpressions(["asset"])
)
table = catalog.createTable(spec)

The fromSource factory reads field types from the Source Type and produces a Lakehouse-friendly schema. See Use Lakehouse Tables in Data Fusion Pipelines.

Add primary keys

Primary keys let merge writes match rows without an explicit on= clause. Declare them at create time:

Python
spec = (
    c3.DataLake.CreateTableSpec.builder()
    .name("turbineMeasurements")
    .schema(schema)
    .primaryKeys(["asset", "t"])
    .build()
)

A primary key is a uniqueness contract enforced by the merge operation. The Data Lakehouse does not reject duplicate inserts; it relies on the merge step to deduplicate.

Add partition expressions

Partitions reduce scan cost. The Data Lakehouse supports four partition styles:

  • Identity: "asset"
  • Time-bucketed: "day(t)", "month(t)", "year(t)", "hour(t)"
  • Hash-bucketed: "bucket(asset, 128)"
  • Composite: ["bucket(asset, 128)", "month(t)"]
Python
spec = (
    c3.DataLake.CreateTableSpec.builder()
    .name("turbineMeasurements")
    .schema(schema)
    .partitionExpressions(["bucket(asset, 128)", "month(t)"])
    .build()
)

Add a sort order

Release 8.11 adds sort order to table creation. A sort order produces files that are sorted by one or more columns. Sorted files reduce work for range queries and ORDER BY queries.

The sort order on CreateTableSpec is a list of DataLake.Table.SortField entries. Build one entry per column you want to sort on:

Python
sort_fields = [
    c3.DataLake.Table.SortField.builder()
        .columnName("t")
        .direction(c3.DataLake.Table.SortDirection.ASC)
        .nullOrder(c3.DataLake.Table.NullOrder.NULLS_LAST)
        .build()
]

spec = (
    c3.DataLake.CreateTableSpec.builder()
    .name("turbineMeasurements")
    .schema(schema)
    .primaryKeys(["asset", "t"])
    .sortOrder(sort_fields)
    .build()
)

Load data from a Spark DataFrame

A Spark DataFrame writes to a table in append, overwrite, or merge mode through Data.SparkSession.

Python
cluster = c3.SparkCluster.inst()
ss = cluster.dataSparkSession()
df = ss.read_csv("gs://atlas-power-sample/turbineMeasurements.csv")
df.write_table(table, mode="append")

For overwrite:

Python
df.write_table(table, mode="overwrite")

For a merge that updates matching rows and inserts new ones:

Python
df.write_table(
    table,
    mode="merge",
    on=["asset", "t"],
    how=(
        c3.MergeActions
        .whenMatched("s.t < '2026-01-01'")
        .update({"activePower": "s.activePower"})
        .whenNotMatched("s.t < '2026-01-01'")
        .insertAll()
    ),
)

A merge can also delete rows that exist in the target but not the source:

Python
how = (
    c3.MergeActions
    .whenMatched().update({"activePower": "s.activePower"})
    .whenNotMatched().insertAll()
    .whenNotMatchedBySource().delete()
)

Load data from a CSV file

For a small or one-time load, importCsv reads a CSV file from a bucket and writes it to the table.

Python
table.importCsv(
    "gs://atlas-power-sample/turbineMeasurements.csv",
    c3.DataLake.Table.ImportMode.APPEND,
    c3.CsvParsingSpec.builder().header(True).build(),
)

ImportMode accepts APPEND and OVERWRITE. There is no merge mode for importCsv; use a Spark DataFrame for merge.

Load data from a Persistable Type

To copy rows from a C3 Persistable Type into a Lakehouse table:

Python
table.upsertBatch(c3.WindTurbineMeasurement.fetchObjStream())

fetchObjStream streams the source rows in memory-bounded batches.

Load data from a Data Fusion Source

To load Source Files that Data Fusion has already synchronized:

Python
c3.DataLake.Operations.writeToDataLake(
    c3.SourceFile.fetch(filter="source=='CanonicalWindTurbineMeasurement'").objs,
    table,
    mode="overwrite",
)

See Use Lakehouse Tables in Data Fusion Pipelines for the full Data Fusion to Lakehouse pattern.

Tips for the Studio Tables tab

The All tables view currently does not list tables in some 8.11 builds. To find a table by name, use the Select Table... combobox in the View Reference Table panel on the SQL Editor tab, which queries the catalog directly.

See also

Was this page helpful?