C3 AI Documentation Home

Transform Data Lakehouse Data with Spark

The C3 AI Data Lakehouse runs on Apache Spark. Spark transforms run from a notebook — the C3 AI Studio Data Lakehouse page does not expose this surface. (For ad-hoc SQL in the browser, use the SQL Editor described in Query Data Lakehouse Tables.) This topic covers the two notebook session flavors — Data.SparkSession for Pandas-on-Spark with C3 helpers (load_table, read_csv, fetch, applyC3Transform), and Spark Connect for native PySpark — and the patterns that bridge Spark and the rest of the C3 Agentic AI Platform.

Choose a session flavor

SessionAPI styleBest for
Data.SparkSessionPandas-on-Spark, with C3 helpers (load_table, read_csv, fetch, applyC3Transform)Most analytics workflows on the C3 Agentic AI Platform.
Spark Connect (cluster.sparkConnectSession())Native PySpark SparkSessionWorkflows that need standard PySpark, including external IDEs and existing Spark code.

You can mix the two. The same underlying Spark cluster runs both.

Start a session

Python
cluster = c3.SparkCluster.inst()
cluster.ensureService(waitForReady=True)

ss = cluster.dataSparkSession()              # Pandas-on-Spark
spark = cluster.sparkConnectSession()        # Native PySpark

To enable execution logging so each call shows up on the Spark Monitoring page:

Python
ss = cluster.dataSparkSession(enableLogging=True)

Read a Lakehouse table

Python
df = ss.load_table(table)

To read by branch or snapshot, see Use Data Lakehouse Branches and Query Data Lakehouse Tables.

Read a C3 Persistable Type into Spark

Use C3TableProvider to read entity data without going through a Lakehouse table.

Python
include = "location,power,manufacturer.name"
spec = c3.Data.SparkFetchSpec.builder().include(include).build()

fetch_df = (
    spark.read.format("c3.spark.c3source.C3TableProvider")
    .option("typeName", "WindTurbine")
    .option("spec", spec.toJsString())
    .option("action", "fetch")
    .load()
)

The include clause follows the same syntax as the Persistable fetch API.

Read a CSV file

Python
df = ss.read_csv("gs://atlas-power-sample/turbineMeasurements.csv")
df.shape

Supported URL schemes match the Data Lakehouse storage prerequisites. See Data Lakehouse Prerequisites.

Apply a C3 Transform to a Spark DataFrame

Transforms in the C3 Agentic AI Platform are reusable mappings from a Source type to a Target type. To run a Transform on a Spark DataFrame:

Python
transform = c3.SourceWindTurbineMeasurement.allTransforms()[0]
target_df = df.applyC3Transform(transform)

The Transform produces a new DataFrame in the shape of the Target type. Use this when you have a Spark DataFrame that holds Source-shaped rows and you want Target-shaped output without leaving Spark.

Write a DataFrame back to a Persistable Type

Python
df.upsert("WindTurbineMeasurement", {"batchSize": 1000})

upsert writes Spark rows into the corresponding Persistable Type. Use this when a Spark transform produces data you want to query through the standard Persistable Type API.

Write a DataFrame to a Lakehouse table

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

For merge semantics and other modes, see Create and Load Data Lakehouse Tables.

Interop between Pandas and Spark

Python
pdf = df.to_pandas()                     # Spark to Pandas
spark_df = df.to_spark()                 # Pandas-on-Spark to native Spark
back = ss.from_spark(spark_df)           # Native Spark to Pandas-on-Spark

to_pandas collects to the driver, so guard against pulling a very large DataFrame.

Supported Pandas-on-Spark APIs

The Pandas-on-Spark coverage for Data.SparkSession matches the upstream PySpark Pandas API for the supported PySpark version, with C3 extensions in c3-specific and date helpers in dt and str. For the full matrix, see:

See also

Was this page helpful?