[SPARK-53498][SDP] Correctly Reference pyspark/pipelines/cli.py from spark-pipelines Binary

### What changes were proposed in this pull request?
When the `spark-pipelines` binary is run, SDP's `cli.py` is submitted to the configured spark cluster via a `spark-submit` request.

Previously, the file path to retrieve `cli.py` was hard-coded with respect to the resolved `SPARK_HOME`. This is incorrect as the package layout is different depending on how spark was installed (ex. Spark tarball installation vs PyPi pyspark installation).

The proposed change is to resolve the file path as a hard coded string with respect to where the pyspark module source code is, rather than `SPARK_HOME`.

### Why are the changes needed?

### Does this PR introduce _any_ user-facing change?

### How was this patch tested?

### Was this patch authored or co-authored using generative AI tooling?

Closes #52254 from AnishMahto/fix-sdp-cli-pyspark.

Authored-by: anishm-db <anish.mahto@databricks.com>
Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

Author: anishm-db

bin/spark-pipelines

@@ -30,4 +30,11 @@ fi
 export PYTHONPATH="${SPARK_HOME}/python/:$PYTHONPATH"
 export PYTHONPATH="${SPARK_HOME}/python/lib/py4j-0.10.9.9-src.zip:$PYTHONPATH"
 
-exec "${SPARK_HOME}"/bin/spark-class org.apache.spark.deploy.SparkPipelines "$@"
+SDP_CLI_PY_FILE_PATH=$("${PYSPARK_PYTHON}" - <<'EOF'
+import pyspark, os
+from pathlib import Path
+print(Path(os.path.dirname(pyspark.__file__)) / "pipelines" / "cli.py")
+EOF
+)
+
+exec "${SPARK_HOME}"/bin/spark-class org.apache.spark.deploy.SparkPipelines "$SDP_CLI_PY_FILE_PATH" "$@"

core/src/main/scala/org/apache/spark/deploy/SparkPipelines.scala

@@ -34,15 +34,16 @@ import org.apache.spark.util.SparkExitCode
  */
 object SparkPipelines extends Logging {
   def main(args: Array[String]): Unit = {
-    val sparkHome = sys.env("SPARK_HOME")
-    SparkSubmit.main(constructSparkSubmitArgs(args, sparkHome).toArray)
+    val pipelinesCliFile = args(0)
+    val sparkSubmitAndPipelinesArgs = args.slice(1, args.length)
+    SparkSubmit.main(
+      constructSparkSubmitArgs(pipelinesCliFile, sparkSubmitAndPipelinesArgs).toArray)
   }
 
   protected[deploy] def constructSparkSubmitArgs(
-      args: Array[String],
-      sparkHome: String): Seq[String] = {
+      pipelinesCliFile: String,
+      args: Array[String]): Seq[String] = {
     val (sparkSubmitArgs, pipelinesArgs) = splitArgs(args)
-    val pipelinesCliFile = s"$sparkHome/python/pyspark/pipelines/cli.py"
     (sparkSubmitArgs ++ Seq(pipelinesCliFile) ++ pipelinesArgs)
   }
 

core/src/test/scala/org/apache/spark/deploy/SparkPipelinesSuite.scala

@@ -35,7 +35,8 @@ class SparkPipelinesSuite extends SparkSubmitTestUtils with BeforeAndAfterEach {
       "spark.conf2=3"
     )
     assert(
-      SparkPipelines.constructSparkSubmitArgs(args, sparkHome = "abc") ==
+      SparkPipelines.constructSparkSubmitArgs(
+        pipelinesCliFile = "abc/python/pyspark/pipelines/cli.py", args = args) ==
       Seq(
         "--deploy-mode",
         "client",
@@ -60,7 +61,8 @@ class SparkPipelinesSuite extends SparkSubmitTestUtils with BeforeAndAfterEach {
       "pipeline.yml"
     )
     assert(
-      SparkPipelines.constructSparkSubmitArgs(args, sparkHome = "abc") ==
+      SparkPipelines.constructSparkSubmitArgs(
+        pipelinesCliFile = "abc/python/pyspark/pipelines/cli.py", args = args) ==
       Seq(
         "--conf",
         "spark.api.mode=connect",
@@ -86,7 +88,8 @@ class SparkPipelinesSuite extends SparkSubmitTestUtils with BeforeAndAfterEach {
       "spark.conf2=3"
     )
     assert(
-      SparkPipelines.constructSparkSubmitArgs(args, sparkHome = "abc") ==
+      SparkPipelines.constructSparkSubmitArgs(
+        pipelinesCliFile = "abc/python/pyspark/pipelines/cli.py", args = args) ==
       Seq(
         "--supervise",
         "--conf",
@@ -109,7 +112,8 @@ class SparkPipelinesSuite extends SparkSubmitTestUtils with BeforeAndAfterEach {
       "org.apache.spark.deploy.SparkPipelines"
     )
     intercept[SparkUserAppException] {
-      SparkPipelines.constructSparkSubmitArgs(args, sparkHome = "abc")
+      SparkPipelines.constructSparkSubmitArgs(
+        pipelinesCliFile = "abc/python/pyspark/pipelines/cli.py", args = args)
     }
   }
 
@@ -120,7 +124,8 @@ class SparkPipelinesSuite extends SparkSubmitTestUtils with BeforeAndAfterEach {
       "myproject"
     )
     assert(
-      SparkPipelines.constructSparkSubmitArgs(args, sparkHome = "abc") ==
+      SparkPipelines.constructSparkSubmitArgs(
+        pipelinesCliFile = "abc/python/pyspark/pipelines/cli.py", args = args) ==
         Seq(
           "--conf",
           "spark.api.mode=connect",

python/packaging/classic/setup.py

@@ -306,6 +306,7 @@ def run(self):
             "pyspark.pandas.spark",
             "pyspark.pandas.typedef",
             "pyspark.pandas.usage_logging",
+            "pyspark.pipelines",
             "pyspark.python.pyspark",
             "pyspark.python.lib",
             "pyspark.testing",