C3 AI Documentation Home

Use Lakehouse Tables in Data Fusion Pipelines

The C3 AI Data Lakehouse is both a source and a target for Data Fusion pipelines. Two of the three integration paths run from the Data Fusion canvas in C3 AI Studio. The third is a notebook path for one-off scripted loads. This topic explains when to use each.

In C3 AI Studio: Data Fusion canvas

Use the Data Fusion canvas when you want a pipeline to read from or write to a Lakehouse table on a schedule, with the same monitoring, lineage, and rerun behavior as the rest of your Data Fusion estate.

Prerequisite: Register the source system and source collection

The Data Fusion canvas can only select a Lakehouse table that already has a registered DataLakeSourceCollection. From a notebook, register the DataLakeSourceCollection and the DataLakeSourceSystem that groups it before you wire the source on the canvas.

The DataLakeSourceSystem holds the catalog and connection settings. Most applications use a single source system named lakehouse that points at the application's default catalog. Create it once:

Python
catalog = c3.DataLake.Catalog.inst()
sourceSystem = (
    c3.DataLakeSourceSystem.forName("lakehouse")
    .withCatalog(catalog)
    .upsert()
)

Each DataLakeSourceCollection wraps a specific DataLake.Table reference and belongs to a source system. Register one per table you want to expose to a pipeline:

Python
table = catalog.table(
    "turbineMeasurements",
    c3.DataLake.Catalog.TableSpec.builder().namespace("dfl").build(),
)

sc = (
    c3.DataLakeSourceCollection.forName("turbineMeasurementsLakehouseSource")
    .withSourceSystem(sourceSystem)
    .withTable(table)
    .upsert()
)

Read a Lakehouse table as a pipeline source

  1. Open your application in C3 AI Studio.
  2. Select the Data Fusion tab in the Data section.
  3. Open the pipeline you want to extend.
  4. Select Add data source and choose Data Lakehouse.
  5. Select the DataLakeSourceCollection you registered above.
  6. Map the source columns to the Source Type fields used by the pipeline.

The pipeline now reads from the Lakehouse table on each run. See Configure the Source System and Source Collection for the general source-collection workflow.

Write pipeline output to a Lakehouse table

Use a Data Fusion Load to Data Lakehouse step when you want a pipeline to write its transformed output to a Lakehouse table.

  1. Open the pipeline in the Data Fusion canvas.
  2. Add a load step downstream of the transform step.
  3. Select Data Lakehouse as the load target.
  4. Select the Catalog and Namespace for the output table.
  5. Enter the table name. If the table does not exist, the pipeline creates it on the first run.
  6. Select the Write mode: Append, Overwrite, or Merge.
  7. Save the pipeline.

The Load to Data Lakehouse step uses DataInteg.Pipeline.Step.Load.OutputConfiguration.DataLake under the hood. The first run creates the table with a schema derived from the upstream step's output. Subsequent runs append, overwrite, or merge based on the configured mode.

Trigger the pipeline from the Data Fusion canvas. The run shows up on the Spark Executions page with Origin = Source to DataLake — the SparkExecutionOrigin value the platform uses for writes that land in a Lakehouse table from a pipeline.

From a notebook: scripted load with writeToDataLake

Use the notebook path when you want to load Source Files directly into a Lakehouse table without a full Data Fusion pipeline. This path is the bridge between SourceFile and the Data Lakehouse, and is the right choice for one-off or scripted loads.

Sync the Source Files

Python
c3.SourceFile.syncAll()
c3.InvalidationQueue.waitForCompute()

syncAll triggers ingestion of any new files; waitForCompute blocks until the resulting compute work finishes.

Read Source Files into a Spark DataFrame

Python
ss = cluster.dataSparkSession()
measurements = ss.read_source_files(
    c3.SourceFile.fetch(
        filter="source=='CanonicalWindTurbineMeasurement'"
    ).objs
)

read_source_files returns a Spark DataFrame shaped like the Source Type.

Create the target table from the Source Type

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

fromSource inherits the schema from the Source Type and turns it into a Lakehouse-friendly column list.

Write the Source Files into the table

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

writeToDataLake reads the listed Source Files and writes them to the table; the DataLake.Table.WriteSpec controls write mode and any merge configuration.

Choose between the three paths

NeedPath
Data Fusion pipeline reads from a Lakehouse tableRead a Lakehouse table as a pipeline source
Data Fusion pipeline output should land in a Lakehouse tableWrite pipeline output to a Lakehouse table
One-off or scripted load from Source Files to a Lakehouse tableScripted load with writeToDataLake

Notes on the Visual Notebooks Delta Lake node

The Visual Notebooks (formerly C3 AI Ex Machina) node library exposes a Delta Lake Output Node. That node targets Azure Delta Lake, a separate product from the C3 AI Data Lakehouse described in this guide. The two formats share a similar transactional model, but the catalogs, APIs, and storage formats are not interchangeable. Use the Data Lakehouse load paths in this topic when your target is the C3 Iceberg Lakehouse.

See also

Was this page helpful?