You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by sz...@apache.org on 2026/08/03 18:10:25 UTC
(spark) branch branch-4.x updated: [SPARK-57594][SDP] Add spark_confs support to create_auto_cdc_flow
This is an automated email from the ASF dual-hosted git repository.
szehon-ho pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 6225f7ebd92d [SPARK-57594][SDP] Add spark_confs support to create_auto_cdc_flow
6225f7ebd92d is described below
commit 6225f7ebd92ddfc323e1603f5eea8c7ada839bf0
Author: trisjain-db <tr...@databricks.com>
AuthorDate: Mon Aug 3 11:09:56 2026 -0700
[SPARK-57594][SDP] Add spark_confs support to create_auto_cdc_flow
### What changes were proposed in this pull request?
Wires per-flow Spark configuration through to AutoCDC flows on the OSS Spark Connect pipelines path, matching the behavior already supported by regular (relation) flows.
- `python/pyspark/pipelines/flow.py`: add `spark_conf: Dict[str, str]` field to the `AutoCdcFlow` dataclass.
- `python/pyspark/pipelines/api.py`: add `spark_conf: Optional[Dict[str, str]] = None` kwarg (last param, backwards-compatible) to `create_auto_cdc_flow` and thread it into the `AutoCdcFlow` constructor with `spark_conf or {}`.
- `python/pyspark/pipelines/spark_connect_graph_element_registry.py`: in `register_auto_cdc_flow`, replace the hard-coded `sql_conf={}` on the outgoing `DefineFlow` proto with `sql_conf=flow.spark_conf`.
### Why are the changes needed?
`pyspark.pipelines.create_auto_cdc_flow` is the only flow-defining API in the `pyspark.pipelines` surface that does not accept a `spark_conf` keyword. The peer APIs all do:
- `append_flow(spark_conf=...)`
- `table(spark_conf=...)`
- `materialized_view(spark_conf=...)`
- `temporary_view(spark_conf=...)`
The `DefineFlow.sql_conf` proto field already exists and is shared by all flow variants, and the Scala server already reads it into `AutoCdcFlow.sqlConf` for use by `FlowAnalysis`. The Python client simply was not populating the field for AutoCDC flows (the registry hard-coded `sql_conf={}`).
Without this kwarg, users of `create_auto_cdc_flow` have no way to apply per-flow Spark configurations. They are forced to set the conf at the pipeline or cluster scope, which then applies to every flow in the pipeline.
This change closes the gap so AutoCDC flows have the same conf-scoping precedence as the other operators: **flow > destination > pipeline > cluster**.
### Does this PR introduce _any_ user-facing change?
AFFECTED VERSIONS: master
`pyspark.pipelines.create_auto_cdc_flow` previously had no way to set per-flow Spark configurations, unlike `append_flow` / `table` / `materialized_view`. This adds a `spark_conf` keyword argument so users can override session/cluster confs per AutoCDC flow at execution time.
### How was this patch tested?
- Extended `test_create_auto_cdc_flow_with_all_args` to assert `spark_conf` propagates onto the registered flow.
- Added `test_create_auto_cdc_flow_spark_conf_defaults_to_empty` to verify the default is `{}` when the kwarg is omitted.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude
Closes #56671 from trisjain-db/add_spark_conf.
Authored-by: trisjain-db <tr...@databricks.com>
Signed-off-by: Szehon Ho <sz...@gmail.com>
(cherry picked from commit 3c28a9c093f1026d76e53d3eb2b846ffb28465c8)
Signed-off-by: Szehon Ho <sz...@gmail.com>
---
python/pyspark/pipelines/api.py | 5 +++++
python/pyspark/pipelines/flow.py | 3 +++
.../spark_connect_graph_element_registry.py | 2 +-
python/pyspark/pipelines/tests/test_auto_cdc_flow.py | 16 ++++++++++++++++
.../sql/connect/pipelines/PythonPipelineSuite.scala | 20 ++++++++++++++++++++
5 files changed, 45 insertions(+), 1 deletion(-)
diff --git a/python/pyspark/pipelines/api.py b/python/pyspark/pipelines/api.py
index 19bd47e4af76..3ab529cc0390 100644
--- a/python/pyspark/pipelines/api.py
+++ b/python/pyspark/pipelines/api.py
@@ -541,6 +541,7 @@ def create_auto_cdc_flow(
*,
track_history_column_list: Optional[Union[List[str], List[Column]]] = None,
track_history_except_column_list: Optional[Union[List[str], List[Column]]] = None,
+ spark_conf: Optional[Dict[str, str]] = None,
) -> None:
"""
Create an Auto CDC flow into the target table from the Change Data Capture (CDC) source.
@@ -600,6 +601,9 @@ def create_auto_cdc_flow(
to be 2.
:param name: The name of the flow for this create_auto_cdc_flow command. When unspecified, \
this will build a "default flow" with name equal to the target name.
+ :param spark_conf: A dict whose keys are the conf names and values are the conf values. \
+ These confs will be set when the flow is executed; they can override confs set for the \
+ destination, for the pipeline, or on the cluster.
"""
# Lazy import: pyspark.sql.connect.functions.builtin transitively imports grpc, which is
# not available in the docs-build environment. pyspark.pipelines.api is loaded eagerly
@@ -736,6 +740,7 @@ def create_auto_cdc_flow(
stored_as_scd_type=stored_as_scd_type,
track_history_column_list=track_history_column_list,
track_history_except_column_list=track_history_except_column_list,
+ spark_conf=spark_conf or {},
source_code_location=source_code_location,
)
diff --git a/python/pyspark/pipelines/flow.py b/python/pyspark/pipelines/flow.py
index 8a0f44e2abe9..237bcf143255 100644
--- a/python/pyspark/pipelines/flow.py
+++ b/python/pyspark/pipelines/flow.py
@@ -65,6 +65,8 @@ class AutoCdcFlow:
history record.
:param track_history_except_column_list: Optional SCD2-only columns excluded from history \
tracking.
+ :param spark_conf: A dict where the keys are the Spark configuration property names and the
+ values are the property values. These properties will be set on the flow.
:param source_code_location: The location of the source code that created this flow.
"""
@@ -79,4 +81,5 @@ class AutoCdcFlow:
stored_as_scd_type: Optional[Literal[1, 2, "1", "2"]]
track_history_column_list: Optional[List[Column]]
track_history_except_column_list: Optional[List[Column]]
+ spark_conf: Dict[str, str]
source_code_location: SourceCodeLocation
diff --git a/python/pyspark/pipelines/spark_connect_graph_element_registry.py b/python/pyspark/pipelines/spark_connect_graph_element_registry.py
index a46eaca5f4cb..2d8f6c34feb9 100644
--- a/python/pyspark/pipelines/spark_connect_graph_element_registry.py
+++ b/python/pyspark/pipelines/spark_connect_graph_element_registry.py
@@ -165,7 +165,7 @@ class SparkConnectGraphElementRegistry(GraphElementRegistry):
flow_name=flow.name,
target_dataset_name=flow.target,
auto_cdc_flow_details=auto_cdc_details,
- sql_conf={},
+ sql_conf=flow.spark_conf,
source_code_location=source_code_location_to_proto(flow.source_code_location),
)
diff --git a/python/pyspark/pipelines/tests/test_auto_cdc_flow.py b/python/pyspark/pipelines/tests/test_auto_cdc_flow.py
index c142e1b1562c..1c57de3ed888 100644
--- a/python/pyspark/pipelines/tests/test_auto_cdc_flow.py
+++ b/python/pyspark/pipelines/tests/test_auto_cdc_flow.py
@@ -72,11 +72,27 @@ class AutoCdcFlowConstructionTest(unittest.TestCase):
column_list=[col("id"), col("val")],
stored_as_scd_type=1,
name="my_flow",
+ spark_conf={"spark.sql.shuffle.partitions": "8"},
)
flow = cast(AutoCdcFlow, registry.auto_cdc_flows[0])
self.assertEqual(flow.name, "my_flow")
self.assertEqual(flow.stored_as_scd_type, 1)
+ self.assertEqual(flow.spark_conf, {"spark.sql.shuffle.partitions": "8"})
+
+ def test_create_auto_cdc_flow_spark_conf_defaults_to_empty(self):
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("target")
+ dp.create_auto_cdc_flow(
+ target="target",
+ source="source",
+ keys=[col("key")],
+ sequence_by=expr("seq"),
+ )
+
+ flow = cast(AutoCdcFlow, registry.auto_cdc_flows[0])
+ self.assertEqual(flow.spark_conf, {})
def test_create_auto_cdc_flow_with_string_args(self):
# Verify that string forms of column / expression arguments are normalized to
diff --git a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala
index c21a4a9ac25b..17ce101bf181 100644
--- a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala
+++ b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala
@@ -1103,6 +1103,26 @@ class PythonPipelineSuite
assert(flow.destinationIdentifier == graphIdentifier("target"))
}
+ test("AutoCDC API: spark_conf is forwarded to the flow's sqlConf") {
+ val flow = buildAutoCdcFlow("""
+ |@dp.table
+ |def src():
+ | return spark.readStream.format("rate").load()
+ |
+ |dp.create_streaming_table("target")
+ |
+ |dp.create_auto_cdc_flow(
+ | target = "target",
+ | source = "src",
+ | keys = ["value"],
+ | sequence_by = "timestamp",
+ | spark_conf = {"spark.sql.shuffle.partitions": "8"},
+ |)
+ |""".stripMargin)
+
+ assert(flow.sqlConf == Map("spark.sql.shuffle.partitions" -> "8"))
+ }
+
test("AutoCDC API: multi-part `keys` column is rejected at flow registration") {
val ex = intercept[RuntimeException] {
buildAutoCdcFlow("""
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org