Spark integration

The cats-eo-spark module turns the Spark SQL operations that are optics into eo optic constructions, so they compose with andThen and run through the usual .get / .modify / .replace / .getOption surface. The operations that are not optics (joins, grouping, explode) become consumers that read their keys through a capability.

The central idea: a Spark Column is a plain Scala value, an unevaluated expression. A lens whose source and focus are typed columns is an ordinary eo Lens. Its get builds a getField, its replace builds a withField, and Spark's optimizer compiles the composed expression. No row is ever decoded into a JVM object.

libraryDependencies ++= Seq(
  "dev.constructive" %% "cats-eo-spark" % "1.0",
  // Spark publishes Scala 2.13 artifacts only; Scala 3 consumes them like this:
  ("org.apache.spark" %% "spark-connect-client-jvm" % "4.2.0").cross(CrossVersion.for3Use2_13),
)

The module compiles against spark-sql-api alone (Provided). That is the API Spark Classic and Spark Connect share.

Runtime: Spark Connect

Spark Classic cannot run in a JVM on Scala 3.8+. Since 3.8 the Scala standard library is compiled by Scala 3, and Spark's Scala 2 runtime reflection (ScalaReflection, the reflective session lookup in SparkSession.builder()) can no longer start on it: class Array does not have a member apply. Use Spark Connect instead. The client runs in your Scala 3 JVM, and the engine runs in a Connect server, on Spark's own Scala 2.13 stdlib.

import org.apache.spark.sql.SparkSession

// NOT org.apache.spark.sql.SparkSession.builder(): its implementation lookup is reflective.
val spark: SparkSession =
  org.apache.spark.sql.connect.SparkSession.builder().remote("sc://localhost:15002").getOrCreate()

Two consequences:

• Encoders come from eo. SparkEncoders.derived derives case-class encoders at compile time (a Hearth macro, like eo-generics'). Spark's Encoders.product needs a Scala 2 TypeTag, and its ProductEncoder decode path needs ScalaReflection.
• Collections and tuples are covered too, through paths Spark decodes without reflection.

import org.apache.spark.sql.catalyst.encoders.AgnosticEncoder
import org.apache.spark.sql.functions.{col, lit, struct, upper}
import dev.constructive.eo.*
import dev.constructive.eo.spark.*
import dev.constructive.eo.spark.SparkEncoders.given

case class Address(street: String, zip: Int)
case class Person(name: String, address: Address, work: Option[Address])

given AgnosticEncoder[Address] = SparkEncoders.derived
given AgnosticEncoder[Person] = SparkEncoders.derived

Typed columns and struct lenses

Col[A] is an opaque Column tagged with the Scala shape of its values. A struct field is a core Lens over Cols. A row is a struct column, so its edits are column expressions too:

val row: Col[Person] = Col(struct(col("name"), col("address"), col("work")))
// row: Col[Person] = struct(name, address, work)

val street = ColumnOptics.field[Person, Address]("address")
  .andThen(ColumnOptics.field[Address, String]("street"))
// street: GetReplaceLens[Col[Person], Col[Person], Col[String], Col[String]] = dev.constructive.eo.optics.GetReplaceLens@62fcfb57

street.get(row).column
// res0: Column = struct(name, address, work)['address']['street']
street.modify(s => Col(upper(s.column)))(row).column
// res1: Column = update_field(struct(name, address, work), address, update_field(struct(name, address, work)['address'], street, upper(struct(name, address, work)['address']['street'])))

The selector macro builds the same optic from a path, and also builds its eager twin, the ordinary optic over Person values. Spark runs the first and Scala runs the second. Because both come from one selector, they address the same field by construction:

val streetPath = ColumnOptics.path[Person](_.address.street)
// streetPath: Total[Person, String] = dev.constructive.eo.spark.ColumnPath$Total@79822ce7

streetPath.symbolic.modify(s => Col(upper(s.column)))(row).column
// res2: Column = update_field(struct(name, address, work), `address`.`street`, upper(struct(name, address, work)['address']['street']))
streetPath.eager.modify(_.toUpperCase)(Person("eo", Address("main st", 1), None))
// res3: Person = Person(
//   name = "eo",
//   address = Address(street = "MAIN ST", zip = 1),
//   work = None
// )

Partial foci: partiality lives in the complement

A symbolic optic runs when the query is built, before any row exists. A partial focus therefore has two kinds of miss, and core Optional holds both. There is no new carrier.

• A plan-time miss (the schema lacks the field) is Left: getOption is None and modify hands the column back unchanged.
• A per-row miss is a plan-time hit whose focus is NULL on the missing rows. The rebuild is guarded by when(present, rebuild).otherwise(source), with the source taken from the complement, so missing rows come out exactly as they went in.

.get on an Option field marks a nullable step:

val workZip = ColumnOptics.path[Person](_.work.get.zip)
// workZip: Partial[Person, Int] = dev.constructive.eo.spark.ColumnPath$Partial@3150f700

workZip.symbolic.modify(z => Col(z.column + 1))(row).column
// res4: Column = CASE WHEN isNotNull(struct(name, address, work)['work']) THEN update_field(struct(name, address, work), `work`.`zip`, +(struct(name, address, work)['work']['zip'], 1)) ELSE struct(name, address, work) END

Catalyst folds that guard into the NULL check withField emits anyway. The optimized plan of a whole-Dataset edit is the projection you would write by hand:

Project [name#16, address#17, if (isnull(work#18)) null else named_struct(street, work#18.street, zip, (work#18.zip + 1)) AS work#31]
+- LocalRelation [name#16, address#17, work#18]

The other partial foci follow the same pattern:

json carries a reserved corrupt-record column because plain PERMISSIVE from_json decodes malformed or non-object input to a struct of NULL fields, not to NULL. A naive presence test would then rewrite "not json" as {}.

Containers: writes are Modify, reads are aggregates

A symbolic array has no elements to hand out while the query is being built, so element-wise writes are a core Modify (transform, transform_values) and element-wise reads are Getters into one aggregate. An effectful modifyA cannot run inside a Spark expression at all.

case class Item(sku: String, qty: Int, price: Double)
case class Order(id: String, items: List[Item])

val orderRow: Col[Order] = Col(struct(col("id"), col("items")))
// orderRow: Col[Order] = struct(id, items)
val qty = ColumnOptics.field[Order, List[Item]]("items")
  .andThen(ColumnOptics.each[List, Item])
  .andThen(ColumnOptics.field[Item, Int]("qty"))
// qty: Optic[Col[Order], Col[Order], Col[Int], Col[Int], [X >: Nothing <: Any, A >: Nothing <: Any] =>> ModifyF[X, A]] = repl.MdocSession$MdocApp$$anon$11@54074694

qty.modify(q => Col(q.column * 2))(orderRow).column
// res5: Column = update_field(struct(id, items), items, transform(struct(id, items)['items'], x_1 -> update_field(x_1, qty, *(x_1['qty'], 2))))

Datasets: one projection, consumed through capabilities

DatasetOptics lifts any Col[S] optic onto a Dataset[S] as one projection, whose wrapping struct Catalyst removes. The non-optic operations take the optic as a capability, per eo's consume-via-capability rule:

import org.apache.spark.sql.{Dataset, SparkSession}
import org.apache.spark.sql.functions.count
import dev.constructive.eo.spark.DatasetOptics.*

val spark: SparkSession = ???
val people: Dataset[Person] = spark.createDataset(Seq.empty[Person])

people.editRows(streetPath.symbolic.modify(s => Col(upper(s.column))))   // DataFrame

given CanModify[Col[Person], Col[String]] = streetPath.symbolic
people.modifyAll[String](s => Col(upper(s.column)))                      // Dataset[Person]

given CanGetOption[Col[Person], Col[Int]] = workZip.symbolic
people.filterHits[Int]                                                   // rows with a work zip

joinOn, groupAggregate and explodeAll follow the same pattern. Each one reads its keys and values through CanGet.

The eager bridge

mapThrough runs an eager optic inside a typed Dataset.map. It is the escape hatch for edits a column expression cannot say. It leaves Catalyst, and the closure runs on the server, so the session needs:

• the Scala 3 runtime, cats and cats-eo (ScalaRuntime.addTo(spark) uploads their jars);
• your application's classes.

Optics are Serializable. Define bridge functions where the server can load their class, e.g. in a top-level object.

import org.apache.spark.sql.{Dataset, SparkSession}
import dev.constructive.eo.spark.DatasetOptics.*

object Fns:
  val upperCase: String => String = _.toUpperCase

val spark: SparkSession = ???
ScalaRuntime.addTo(spark) // plus spark.addArtifact("my-app.jar")

given CanModify[Person, String] = streetPath.eager
spark.createDataset(Seq.empty[Person]).mapThrough[String](Fns.upperCase)

The cost, for one nested edit over a cached 200 000-row Dataset (local Connect server, JDK 25, sparkBench):

The bridge costs what any typed Dataset.map costs: every row is decoded to JVM objects and re-encoded. The optic itself adds nothing measurable.

Spark's own values: schemas, rows and variants

These kits need no session. Schemas covers the DataType tree. It is not named DataTypes, because Spark has a Java factory of that name in org.apache.spark.sql.types. Plated gives whole-schema rewrites:

import org.apache.spark.sql.types.*
import dev.constructive.eo.optics.Plated
import Schemas.platedDataType

val schema: DataType = StructType.fromDDL("id INT, address STRUCT<zip: INT, street: STRING>")

Schemas.field("address").andThen(Schemas.field("zip")).getOption(schema)
// res8: Option[DataType] = Some(IntegerType)
Plated.transform[DataType] { case IntegerType => LongType; case t => t }(schema).simpleString
// res9: String = "struct<id:bigint,address:struct<zip:bigint,street:string>>"

Rows gives typed field[A] / at[A] access into untyped Rows, keeping the schema on writes. Variants edits Spark 4 VARIANT values. The binary format has no in-place update, so writes rebuild the value, always in one canonical encoding. That makes VariantVal's byte equality structural:

val doc = Variants.obj(
  "name" -> Variants.string.reverseGet("eo"),
  "tags" -> Variants.arr(Variants.long.reverseGet(1L), Variants.long.reverseGet(2L)),
)
// doc: VariantVal = {"name":"eo","tags":[1,2]}

val tags = Variants.field("tags").andThen(Variants.each).andThen(Variants.long)
// tags: Optic[VariantVal, VariantVal, Long, Long, [X >: Nothing <: Any, A >: Nothing <: Any] =>> MultiFocusK[[B >: Nothing <: Any] =>> PSVec[B], X, A]] = repl.MdocSession$MdocApp$$anon$19@d0584ed
tags.modify(_ * 10)(doc).toJson(java.time.ZoneOffset.UTC)
// res10: String = "{\"name\":\"eo\",\"tags\":[10,20]}"

Which Spark operations are optics?