Apache Hudi through Spark Connect: running the full table surface from a thin client

• Architecture: what Spark Connect changes for Hudi
• Starting a server that speaks Hudi
  • The extension class name, which is the usual mistake
• Connecting the client
• Copy-on-write tables, the full DML surface
  • The partitioning trap
  • Update, merge and delete
• Schema evolution: two of four operations
• Merge-on-read, and what the two query types show
• Incremental queries and time travel
• Procedures and metadata, over gRPC
  • rollback_to_instant only unwinds the newest instant
• What the thin client cannot do
• Production notes
• Frequently asked questions
• Conclusion
• References
• Trademarks

TL;DR

• Spark Connect moves planning to a server, so the Hudi bundle, spark.sql.extensions, the Kryo serializer and HoodieCatalog all belong to the server at startup. A client cannot add them later.
• The client becomes a pip install with no Spark, no Java and no Hudi jar. Everything below was driven from one.
• Almost the whole surface works: COW and MOR, partitioned tables, INSERT, UPDATE, MERGE, DELETE, incremental queries, time travel, 12 stored procedures and the metadata table functions.
• Four things do not. RENAME COLUMN and DROP COLUMN are rejected outright, rollback_to_instant only unwinds the newest instant, and hudi_table_changes in CDC mode needs a CDC-enabled table.
• The trap that costs the most time is not a Connect issue at all: partitioning moves the partition column to the end of the schema, so a positional INSERT fails with an error that blames the wrong column.

Hudi is not part of Spark. It arrives as a bundle jar, installs a Catalyst extension, registers a catalog and needs a specific serializer. All four are session-level concerns, and Spark Connect moves the session to the far side of a gRPC connection.

So the practical question is not whether Hudi works through Spark Connect. It is which half of your setup goes where, and what the failures look like when a piece lands on the wrong side. This post answers that by running the whole thing: one Connect server, one client with nothing installed, and every Hudi operation worth using, including the ones that fail.

Every result quoted here came out of that run. Where something failed I say so, and where a failure turned out to be my own mistake rather than a limitation I say that too, because the error messages in two of those cases point somewhere misleading.

Architecture: what Spark Connect changes for Hudi

In a classic session your process owns the SparkSession, the analyzer, the catalog and the scheduler. Hudi’s extension and catalog are installed into your session, by your own spark-submit line.

Spark Connect cuts that in half:

flowchart TB
  subgraph C["client, your laptop"]
    A["DataFrame and SQL calls"] --> B["unresolved plan<br/>as protobuf"]
  end
  B -->|"gRPC, port 15002"| S
  subgraph S["Spark Connect server, the driver"]
    D["analyzer plus HoodieCatalog"] --> E["Catalyst optimiser plus<br/>HoodieSparkSessionExtension"]
    E --> F["physical plan"]
  end
  F --> X["executors, Hudi bundle on the classpath"]
  X -->|"Arrow batches"| A

The client cannot resolve default.trips_cow because it has no catalog and no idea what a Hudi table is. It sends the operation and lets the server work it out. Everything below follows from that single split.

Hudi plugs into Spark at four points, and Connect puts all four on the server:

flowchart TB
  subgraph CL["client"]
    P["pip install pyspark<br/>no Java, no Hudi jar"]
  end
  subgraph SV["server, fixed at startup"]
    J["hudi-spark4.0-bundle"]
    K["HoodieSparkSessionExtension"]
    L["HoodieCatalog over spark_catalog"]
    M["KryoSerializer"]
  end
  P -->|"SQL and DataFrame"| SV
  J --- K --- L --- M

None of the four can be set by a client. spark.conf.set reaches the server for runtime SQL configs, but spark.sql.extensions is read once while the session is built, before any client connects. A client can read it, which is the fastest way to tell a broken server from a broken query.

Starting a server that speaks Hudi

Use --packages when the host can reach Maven Central, --jars with pre-downloaded files when it cannot. Both were run for this post.

$SPARK_HOME/sbin/start-connect-server.sh \
  --packages org.apache.hudi:hudi-spark4.0-bundle_2.13:1.2.0 \
  --conf spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog \
  --conf spark.sql.warehouse.dir=/tmp/warehouse/spark \
  --master "local[4]"

The coordinate encodes two compatibilities at once, and both are hard failures rather than warnings:

Spark 4 is Scala 2.13 only, so a _2.12 bundle never loads.

The extension class name, which is the usual mistake

Hudi’s session extension is org.apache.spark.sql.hudi.HoodieSparkSessionExtension. It lives under org.apache.spark.sql, not under org.apache.hudi, which is where most people reach for it first, and where I reached for it first.

Getting it wrong does not stop the server. That is what makes it expensive: the server starts, reports healthy, and then every MERGE and CALL fails later with an error about syntax or a missing table. The only evidence is a warning in the server log at startup:

WARN SparkSession: Cannot use org.apache.hudi.HoodieSparkSessionExtension to configure session extensions.
java.lang.ClassNotFoundException: org.apache.hudi.HoodieSparkSessionExtension

Two habits remove this whole class of problem. Read the name out of the jar rather than from memory:

unzip -l hudi-spark4.0-bundle_2.13-1.2.0.jar | grep SessionExtension
#   11028  org/apache/spark/sql/hudi/HoodieSparkSessionExtension.class

And confirm from the client that the server loaded it:

print(spark.conf.get("spark.sql.extensions"))
# org.apache.spark.sql.hudi.HoodieSparkSessionExtension

That one line separates “my SQL is wrong” from “the server is misconfigured”, and it is the first thing to run when a MERGE is rejected.

Connecting the client

The client needs pyspark and gRPC. No Java, no SPARK_HOME, no Hudi jar. Pin it to the version your server runs:

pip install "pyspark[connect]==4.0.2"

from pyspark.sql import SparkSession

spark = SparkSession.builder.remote("sc://localhost:15002").getOrCreate()
print(spark.version)

spark.version reports the server’s version, so it tells you nothing about your client. Check the client separately with pyspark.__version__.

The pin matters more than it looks, and a bare pip install "pyspark[connect]" is the wrong instruction. It installs whatever is newest on PyPI, and a client newer than the server fails on the first createDataFrame:

SparkNoSuchElementException: [SQL_CONF_NOT_FOUND] The SQL config
"spark.sql.session.localRelationSizeLimit" cannot be found. SQLSTATE: 42K0I

The newer client asks the older server for a SQL config that does not exist there. Every SQL statement in this post still works in that configuration, so the failure looks arbitrary: DDL, MERGE, procedures and metadata queries all succeed, and only the paths that build a local relation break.

The gap is not symmetric. An older client against a newer server was fine: a 3.5.9 client drove the same 4.0.2 server through every example here, createDataFrame included. A newer client against an older server was not. Match the versions, and when you cannot, be the older side.

Copy-on-write tables, the full DML surface

Hudi installs over the default catalog rather than adding a named one, so tables are ordinary two-part names. The Hudi specifics go in TBLPROPERTIES.

spark.sql("""
  CREATE TABLE default.trips_cow (
    trip_id    BIGINT,
    rider_id   BIGINT,
    city       STRING,
    fare       DECIMAL(10,2),
    started_at TIMESTAMP)
  USING hudi
  PARTITIONED BY (city)
  TBLPROPERTIES (
    primaryKey      = 'trip_id',
    preCombineField = 'started_at',
    type            = 'cow')
  LOCATION '/tmp/warehouse/trips_cow'""")

primaryKey is what makes an UPDATE or MERGE possible at all, since Hudi needs a record key to find the row it is replacing. preCombineField decides which version wins when two records share a key. type = 'cow' selects copy-on-write, where an update rewrites the whole base file, as opposed to merge-on-read, which appends a delta log and merges at query time.

The partitioning trap

Now the thing that will cost you an afternoon. This insert looks obviously correct and fails:

spark.sql("""
  INSERT INTO default.trips_cow VALUES
    (1, 101, 'Hyderabad', 245.50, TIMESTAMP '2026-09-01 08:15:00')""")

AnalysisException: [INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_SAFELY_CAST]
Cannot write incompatible data for the table `spark_catalog`.`default`.`trips_cow`:
Cannot safely cast `fare` "STRING" to "DECIMAL(10,2)". SQLSTATE: KD000

The error names fare, and fare is fine. Look at the schema Hudi actually created:

[f.name for f in spark.table("default.trips_cow").schema.fields
 if not f.name.startswith("_hoodie")]

['trip_id', 'rider_id', 'fare', 'started_at', 'city']

Partitioning moves the partition column to the end of the schema. The table was declared with city third, and it is stored fifth. A positional INSERT ... VALUES maps by position, so 'Hyderabad' lands on fare, and the cast error is a truthful report of a problem one column to the left of where you are looking.

Two fixes. Name the columns, which is worth doing on any Hudi table whether it is partitioned or not:

spark.sql("""
  INSERT INTO default.trips_cow (trip_id, rider_id, city, fare, started_at) VALUES
    (1, 101, 'Hyderabad', 245.50, TIMESTAMP '2026-09-01 08:15:00'),
    (2, 102, 'Bengaluru', 180.00, TIMESTAMP '2026-09-01 09:30:00'),
    (3, 103, 'Hyderabad', 320.75, TIMESTAMP '2026-09-02 07:45:00')""")

Or supply the values in Hudi’s stored order, with the partition column last. Both were verified; the explicit column list is the one to use, because it survives a later ADD COLUMNS.

Update, merge and delete

spark.sql("UPDATE default.trips_cow SET fare = 250.00 WHERE trip_id = 1")

spark.sql("""
  MERGE INTO default.trips_cow t
  USING (SELECT 4L AS trip_id, 104L AS rider_id, 'Pune' AS city,
                CAST(99.00 AS DECIMAL(10,2)) AS fare,
                TIMESTAMP '2026-09-03 10:00:00' AS started_at) s
    ON t.trip_id = s.trip_id
  WHEN MATCHED     THEN UPDATE SET *
  WHEN NOT MATCHED THEN INSERT *""")

spark.sql("DELETE FROM default.trips_cow WHERE trip_id = 2")

All three are extension-provided syntax, so they are also a working test that the server loaded HoodieSparkSessionExtension. Results from the sequence above:

count after INSERT                      3
fare for trip_id 1 after UPDATE         250.00
count after MERGE (Pune inserted)       4
count after DELETE                      3
count WHERE city = 'Hyderabad'          2
SHOW PARTITIONS   city=Bengaluru, city=Hyderabad, city=Pune

SHOW PARTITIONS still listing city=Bengaluru after its only row was deleted is correct: the partition directory and its metadata remain until cleaning removes them.

The five Hudi bookkeeping columns come through the client untouched, which is a quick way to confirm you are reading a real Hudi table and not a directory of Parquet:

[c for c in spark.table("default.trips_cow").columns if c.startswith("_hoodie")]

['_hoodie_commit_time', '_hoodie_commit_seqno', '_hoodie_record_key',
 '_hoodie_partition_path', '_hoodie_file_name']

Schema evolution: two of four operations

ADD COLUMNS works, and old rows read back NULL for the new column as they should:

spark.sql("ALTER TABLE default.trips_cow ADD COLUMNS (pay_method STRING)")

spark.sql("""
  INSERT INTO default.trips_cow (trip_id, rider_id, city, fare, started_at, pay_method)
  VALUES (5, 105, 'Chennai', 75.25, TIMESTAMP '2026-09-04 11:00:00', 'upi')""")

pay_method for trip_id 5           upi
rows WHERE pay_method IS NULL      3

Widening a column type with ALTER COLUMN ... TYPE is accepted too. The other two are not:

spark.sql("ALTER TABLE default.trips_cow RENAME COLUMN pay_method TO pm")
spark.sql("ALTER TABLE default.trips_cow DROP COLUMN pay_method")

AnalysisException: [UNSUPPORTED_FEATURE.TABLE_OPERATION] The feature is not supported:
Table `spark_catalog`.`default`.`trips_cow` does not support RENAME COLUMN.

AnalysisException: [UNSUPPORTED_FEATURE.TABLE_OPERATION] The feature is not supported:
Table `spark_catalog`.`default`.`trips_cow` does not support DROP COLUMN

Plan schema changes as additive. A rename means writing a new column and backfilling it, so decide column names before the table has data you care about.

Merge-on-read, and what the two query types show

Merge-on-read is where Hudi’s read path becomes visible, and the divergence is easy to demonstrate through a thin client:

spark.sql("""
  CREATE TABLE default.trips_mor (
    trip_id BIGINT, rider_id BIGINT, city STRING,
    fare DECIMAL(10,2), started_at TIMESTAMP)
  USING hudi
  TBLPROPERTIES (primaryKey = 'trip_id', preCombineField = 'started_at', type = 'mor')
  LOCATION '/tmp/warehouse/trips_mor'""")

spark.sql("""
  INSERT INTO default.trips_mor VALUES
    (1, 101, 'Hyderabad', 245.50, TIMESTAMP '2026-09-01 08:15:00'),
    (2, 102, 'Bengaluru', 180.00, TIMESTAMP '2026-09-01 09:30:00')""")

spark.sql("UPDATE default.trips_mor SET fare = 999.99 WHERE trip_id = 1")

Now read the same row two ways:

# snapshot, the default: base files merged with the delta logs
spark.sql("SELECT fare FROM default.trips_mor WHERE trip_id = 1").collect()[0][0]

# read-optimised: base files only, ignoring the logs
(spark.read.format("hudi")
      .option("hoodie.datasource.query.type", "read_optimized")
      .load("/tmp/warehouse/trips_mor")
      .filter("trip_id = 1").select("fare").collect()[0][0])

snapshot         999.99
read_optimized   245.50

That gap is the whole merge-on-read trade, measured: the snapshot read pays a merge to be correct, the read-optimised read is cheaper and stale until compaction. Both paths work identically through Connect.

Compaction is where expectations need managing. run_compaction is not a command that compacts on demand:

spark.sql("CALL run_compaction(op => 'scheduleandexecute', table => 'default.trips_mor')")

0 rows returned, and CALL show_compaction(...) reports 0 pending compactions

Nothing was scheduled because the delta-commit threshold was not met, which on these defaults takes more delta commits than two updates produce. Setting hoodie.compact.inline.max.delta.commits to 1 at table creation went the other way: compaction then ran inline on the write itself, and the read-optimised query returned 999.99 immediately, with still nothing left for run_compaction to do. Either way the procedure returning zero rows is not a failure, and the way to verify compaction is to watch the read-optimised query catch up, not to read the procedure’s return value.

Incremental queries and time travel

This is the Hudi capability with no direct equivalent in the other formats, and it survives the trip through gRPC intact. The DataFrame writer path, where the hoodie.* options are explicit:

from decimal import Decimal
from pyspark.sql import functions as F

trips = (spark.createDataFrame(
        [(7, 107, "Kochi", Decimal("55.00"), "2026-09-06 10:00:00"),
         (8, 108, "Kochi", Decimal("65.00"), "2026-09-06 11:00:00")],
        "trip_id bigint, rider_id bigint, city string, "
        "fare decimal(10,2), started_at string")
    .withColumn("started_at", F.to_timestamp("started_at")))

options = {
    "hoodie.table.name": "trips_df",
    "hoodie.datasource.write.recordkey.field": "trip_id",
    "hoodie.datasource.write.precombine.field": "started_at",
}

(trips.write.format("hudi")
      .options(**options, **{"hoodie.datasource.write.operation": "bulk_insert"})
      .mode("overwrite").save("/tmp/warehouse/trips_df"))

The three write operations behave as their names promise, which is worth checking once rather than assuming. First an upsert of a row that already exists:

changed = (spark.createDataFrame(
        [(7, 107, "Kochi", Decimal("77.77"), "2026-09-06 12:00:00")],
        "trip_id bigint, rider_id bigint, city string, "
        "fare decimal(10,2), started_at string")
    .withColumn("started_at", F.to_timestamp("started_at")))

# upsert matches on the record key, so the row count does not grow
(changed.write.format("hudi")
        .options(**options, **{"hoodie.datasource.write.operation": "upsert"})
        .mode("append").save("/tmp/warehouse/trips_df"))

Then the same two original rows again, this time with insert:

# insert does NOT match on the record key
(trips.write.format("hudi")
      .options(**options, **{"hoodie.datasource.write.operation": "insert"})
      .mode("append").save("/tmp/warehouse/trips_df"))

after bulk_insert of 2 rows              2 rows
after upsert of trip_id 7                2 rows, fare now 77.77
after insert of the same 2 rows again    4 rows

The last line is the one to notice. insert does not deduplicate on the record key, so re-sending rows you already have doubles them. upsert is what most pipelines want, and it is not what every path defaults to.

Time travel reads a table as of an instant:

commits = [r[0] for r in spark.read.format("hudi")
           .load("/tmp/warehouse/trips_df")
           .select("_hoodie_commit_time").distinct()
           .orderBy("_hoodie_commit_time").collect()]

spark.read.format("hudi").option("as.of.instant", commits[0]) \
     .load("/tmp/warehouse/trips_df").count()

commit times            ['20260918093803909', '20260918093804944', '20260918093805997']
as.of.instant, first    2
current                 4

Two rows against four is time travel working over a gRPC connection. And the incremental read, which returns only what changed after an instant:

(spark.read.format("hudi")
      .option("hoodie.datasource.query.type", "incremental")
      .option("hoodie.datasource.read.begin.instanttime", "0")
      .load("/tmp/warehouse/trips_df")
      .count())

Passing 0 means “everything”, which is how you check the plumbing before wiring a real watermark. In production you store the last processed _hoodie_commit_time and pass that instead. The commit times are timestamps, so yours will differ from the ones above; every other number here is reproducible from a clean warehouse.

Procedures and metadata, over gRPC

People assume the CALL interface will not survive Spark Connect. It does. These all ran from the thin client:

spark.sql("CALL show_commits(table => 'default.trips_cow', limit => 10)").count()
spark.sql("CALL stats_wa(table => 'default.trips_cow')").count()
spark.sql("CALL run_clean(table => 'default.trips_cow')").count()

The table-valued functions work as well, with one argument requirement that is easy to miss:

spark.sql("SELECT count(*) FROM hudi_metadata('default.trips_cow')").collect()
spark.sql("SELECT count(*) FROM hudi_table_changes('default.trips_cow','latest_state','earliest')").collect()
spark.sql("SELECT count(*) FROM hudi_filesystem_view('default.trips_cow','city=Hyderabad')").collect()

hudi_filesystem_view takes a partition as its second argument. Called with only a table name it returns zero rows rather than an error, which reads like an empty table:

hudi_filesystem_view('default.trips_cow')                    0
hudi_filesystem_view('default.trips_cow','city=Hyderabad')   3
hudi_filesystem_view('default.trips_mor','')                 1

For a non-partitioned table pass the empty string. And hudi_table_changes in cdc mode needs a table created with CDC enabled:

IllegalArgumentException: It isn't a CDC hudi table on file:/tmp/warehouse/trips_cow

That is a table-property problem, not a Connect one. Enable hoodie.table.cdc.enabled at creation if you want the CDC feed.

rollback_to_instant only unwinds the newest instant

This one deserves its own note because the error is opaque and my first explanation for it was wrong. Rolling back the newest instant works:

instants = sorted(r[0] for r in
    spark.sql("CALL show_commits(table => 'default.trips_cow', limit => 30)").collect())

spark.sql("CALL rollback_to_instant(table => 'default.trips_cow', "
          "instant_time => '%s')" % instants[-1]).collect()
# [Row(rollback_result=True)], and the row count drops by that commit

Rolling back an older instant fails:

UnknownException: (org.apache.hudi.exception.HoodieRollbackException)
Failed to rollback file:///tmp/warehouse/trips commits 20260918094116830

I first assumed a savepoint on the table was blocking it, and tested that: creating a savepoint and then rolling back the newest instant still succeeded, so the savepoint was not the cause. The actual rule is simpler. Rollback unwinds the tip of the timeline, one commit at a time. It is not “restore to this point”, and the way to go further back is repeated rollbacks or a savepoint plus restore.

The practical mistake this invites is an ordering one. show_commits returns instants newest first, so the first row it hands you is the newest, while sorting those instants ascending puts the oldest first. Sorting a list of Hudi instants and passing the front of it gives rollback_to_instant exactly the argument it cannot process, which is how I produced the failure above in the first place.

What the thin client cannot do

The limitations are narrow and all follow from the client not having a JVM.

spark.sparkContext

PySparkAttributeError: [JVM_ATTRIBUTE_NOT_SUPPORTED] Attribute `sparkContext` is not
supported in Spark Connect as it depends on the JVM.

spark.range(10).rdd.map(lambda r: r.id).collect()

PySparkNotImplementedError: [NOT_IMPLEMENTED] rdd is not implemented.

Neither matters much for table work, since Hudi is driven through SQL and the DataFrame API anyway. What does bite is subtler. A Python float for a decimal column fails with an empty error, where a classic session coerces it:

spark.createDataFrame(
    [(9, 109, "Kochi", 75.25, "2026-09-05 12:00:00")],
    "trip_id bigint, rider_id bigint, city string, "
    "fare decimal(10,2), started_at string")

AssertionError

That is the entire message. The traceback points at convert_decimal in pyspark/sql/conversion.py, which does assert isinstance(value, decimal.Decimal). Pass Decimal("75.25") and it works. Any Hudi table with a money column hits this on the first createDataFrame.

Production notes

• Treat the server config as a contract. The jar, the extension, the catalog and Kryo are fixed at startup, so a change means a restart. Nobody’s client can work around a server that was started wrong.
• Check the extension before debugging SQL. spark.conf.get("spark.sql.extensions") from the client, and grep "Cannot use" on the server log, resolve most “MERGE is not supported” reports.
• Always name columns in INSERT. It costs one line and removes the partition-reordering trap permanently, including after a later ADD COLUMNS.
• Design schema changes as additive. RENAME COLUMN and DROP COLUMN are rejected, so a rename is a new column plus a backfill.
• Use upsert, not insert, unless you mean to duplicate. The insert operation does not deduplicate on the record key.
• Do not judge compaction by the procedure’s return value. run_compaction returning zero rows usually means the delta-commit threshold was not met. Watch the read-optimised query instead.
• Pass Decimal for decimal columns in createDataFrame, and prefer CAST(... AS DECIMAL(10,2)) inside SQL.
• Keep Kryo set even on a server you think is Iceberg-only, because one shared server usually ends up serving both.
• Think twice before adding Hudi’s extension to a server that serves Iceberg. It breaks Iceberg’s SQL time travel, one-directionally, and nothing warns you at startup.

Frequently asked questions

Does the client need the Hudi jar? No, and that is the main reason to use Connect for Hudi work. The client sends a plan naming a table and the server resolves it. A virtualenv with pyspark is the whole client-side dependency, which also removes the jar-version skew that usually makes onboarding painful.

Can one server serve Hudi and Iceberg together? Mostly, and the cost falls entirely on Iceberg. spark.sql.extensions takes a comma-separated list, Hudi takes over spark_catalog while Iceberg gets a named catalog, and both formats’ DDL and DML work. But loading Hudi’s extension alongside Iceberg’s breaks Iceberg’s SQL time travel, which then fails with [INTERNAL_ERROR] Found the unresolved operator: 'TimeTravelRelation. Hudi is unaffected in that arrangement: its own DML, procedures, incremental reads and as.of.instant time travel all keep working. The measurement and the workaround are in Apache Iceberg through Spark Connect.

Why does my MERGE fail with a parse error? Because the session extension is not loaded on the server, from either a wrong class name or a jar missing from the server classpath. MERGE and UPDATE on Hudi are extension-provided syntax, so without it the statement is not merely unsupported, it is unparseable.

Can I set hoodie.* options per job from the client? Write-time hoodie.* options travel fine, because they are passed with the write rather than baked into the session. What you cannot change from a client is the session-level set: the extension, the catalog and the serializer.

Is the partition-column reordering a Spark Connect bug? No. It is how Hudi stores a partitioned table, and a classic session behaves the same way. Connect only makes it more visible, because the error arrives without a local stack trace to poke at.

Do client and server versions have to match? Match them. An older client is tolerated: a 3.5.9 client drove the 4.0.2 server through every example here. A newer one is not, and the failure is specific rather than general. A pip-installed 4.2.0 client ran every SQL statement against the 4.0.2 server and then failed on createDataFrame with SQL_CONF_NOT_FOUND for spark.sql.session.localRelationSizeLimit, because it expects a config the older server does not have. That is why the install line above is pinned.

Conclusion

The useful reframing is that Spark Connect forces a separation which was always there. Hudi is infrastructure: a jar, a Catalyst extension, a catalog and a serializer. Your query is not. Classic Spark let you blur the two by shipping jars with every submit, and most --packages lines in most runbooks are that blur made permanent.

Connect makes you decide once, on the server, and then hands every client a clean API with nothing to install. The cost is exactly the flexibility you were abusing: no per-job jars, no per-job extensions, no sparkContext escape hatch. For a team sharing a Hudi lakehouse, that is a good trade.

What is worth carrying away is how few of the failures were about Spark Connect. An extension class under a package that looks wrong but is not, a partition column silently moved to the end of the schema, insert where upsert was meant, rollback_to_instant that only unwinds the tip, and a float where a Decimal belongs. Every one of those behaves identically in a classic session. Reproduce a failure locally before blaming the transport, which is the habit that turned two of my own confident wrong answers here into the right ones.

References

• Spark Connect overview for the architecture and the client APIs
• Hudi Spark SQL DDL and SQL procedures for table properties and the CALL surface
• Hudi configuration reference for the hoodie.* write and read options used above
• conversion.py, whose convert_decimal is the assertion behind the empty AssertionError
• Apache Hudi architecture for the timeline, file slices and compaction machinery underneath all of this

Trademarks

Apache Spark, Apache Hudi, Apache Iceberg, Apache Hive, Apache Parquet and Apache are either registered trademarks or trademarks of The Apache Software Foundation in the United States and other countries.