如何在PySpark中调用Scala代码执行子流程？

我之前在实际项目里也碰到过类似的PySpark调用Scala子流程的需求，下面几种方法都是经过实践验证的，你可以根据自己的场景来选：

1. 将Scala子流程打包成Jar，通过PySpark的JVM桥接调用 #

这是最常用也是最推荐的方式，能直接复用Spark上下文，性能最优。

步骤： #

• 首先把你的Scala子流程代码打包成可执行Jar包：用sbt或者Maven构建，必须确保依赖的Spark版本和你的PySpark环境完全一致（比如PySpark用3.3.0，Scala代码就要基于Spark 3.3.0编译）。
• 启动PySpark的时候，通过--jars参数引入这个Jar包，或者在代码里用spark.sparkContext.addJar("/path/to/your/scala.jar")动态添加。

两种调用场景： #

场景A：Scala实现DataFrame处理逻辑

比如你的Scala代码是一个处理DataFrame的工具类：

package com.example

import org.apache.spark.sql.DataFrame

object ScalaDataProcessor {
  def process(df: DataFrame): DataFrame = {
    // 这里写你的子流程逻辑：过滤、聚合、复杂转换等
    df.filter("age > 18").groupBy("gender").count()
  }
}

在PySpark里，通过Spark的JVM直接调用这个静态方法，再把返回的Java DataFrame转成PySpark DataFrame：

# 获取Scala对象的JVM引用
scala_processor = spark._jvm.com.example.ScalaDataProcessor

# 把PySpark DataFrame转成Java DataFrame，调用Scala方法
java_df = scala_processor.process(df._jdf)

# 转回PySpark DataFrame继续处理
result_df = spark.createDataFrame(java_df)
result_df.show()

场景B：Scala实现自定义UDF

如果子流程是某个特定计算逻辑，可以写成Scala UDF，在PySpark中注册使用：

package com.example

import org.apache.spark.sql.functions.udf

object ScalaUdfs {
  val calculateScore = udf((value: Int, weight: Double) => {
    // 自定义计算逻辑
    value * weight * 1.2
  })
}

PySpark中注册并调用：

# 注册Scala UDF到PySpark环境
spark.udf.register("calculate_score", spark._jvm.com.example.ScalaUdfs.calculateScore)

# 在DataFrame中使用该UDF
df.withColumn("score", expr("calculate_score(value, weight)")).show()

2. 用Spark Pipeline混合语言组件 #

如果你的业务流程是Spark Pipeline形式，可以把Scala子流程写成自定义的Transformer或Estimator，打包成Jar后，在PySpark中直接加载并加入Pipeline。

比如Scala写一个自定义Transformer：

package com.example

import org.apache.spark.ml.Transformer
import org.apache.spark.ml.param.ParamMap
import org.apache.spark.sql.{DataFrame, Dataset}
import org.apache.spark.sql.types.StructType

class CustomScalaTransformer extends Transformer {
  override def transform(dataset: Dataset[_]): DataFrame = {
    // 子流程逻辑：给数据集添加处理后的新列
    dataset.withColumn("processed", dataset("raw_value").cast("double") * 2)
  }

  override def copy(extra: ParamMap): Transformer = defaultCopy(extra)
  override def uid: String = "custom-scala-transformer"
  override def transformSchema(schema: StructType): StructType = schema.add("processed", "double")
}

PySpark中使用这个Transformer构建Pipeline：

from pyspark.ml import Pipeline
from pyspark.ml.feature import VectorAssembler

# 加载Scala的Transformer并转成PySpark可识别的对象
scala_transformer = spark._jvm.com.example.CustomScalaTransformer()
py_scala_transformer = scala_transformer.asInstanceOf[Transformer]

# 构建混合语言的Pipeline
pipeline = Pipeline(stages=[
    VectorAssembler(inputCols=["value"], outputCol="features"),
    py_scala_transformer
])

# 运行Pipeline并获取结果
model = pipeline.fit(df)
result_df = model.transform(df)

3. 独立进程调用（适合无Spark上下文依赖的Scala脚本） #

如果你的Scala子流程不需要依赖Spark上下文，是一个独立的脚本程序，那可以直接在PySpark中用subprocess模块调用，等待脚本执行完成后再继续。

示例：

import subprocess

# 调用Scala脚本，等待执行完成（check=True会在脚本出错时抛出异常）
subprocess.run(
    ["scala", "/path/to/your/independent_scala_script.scala"],
    check=True,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)

# 脚本执行完成后，读取输出文件继续PySpark流程
df = spark.read.csv("/path/to/output_from_scala.csv")

注意事项 #

• 版本兼容性：Scala版本（比如2.12）和Spark版本必须和PySpark环境完全匹配，否则会出现类加载错误。
• 依赖管理：如果Scala Jar有额外依赖，要确保这些依赖也能被PySpark加载（可以用--packages参数或者把依赖Jar一起加入）。
• 序列化问题：在PySpark和Scala之间传递数据时，尽量用DataFrame（基于Spark内置序列化），避免传递自定义对象，减少序列化异常。

内容的提问来源于stack exchange，提问作者Also