[SPARK-48970][PYTHON][ML] Avoid using SparkSession.getActiveSession in spark ML reader/writer

[WeichenXu123]

What changes were proposed in this pull request?

SparkSession.getActiveSession is thread-local session, but spark ML reader / writer might be executed in different threads which causes SparkSession.getActiveSession returning None.

Why are the changes needed?

It fixes the bug like:

        spark = SparkSession.getActiveSession()
>       spark.createDataFrame(  # type: ignore[union-attr]
            [(metadataJson,)], schema=["value"]
        ).coalesce(1).write.text(metadataPath)
E       AttributeError: 'NoneType' object has no attribute 'createDataFrame'

Does this PR introduce any user-facing change?

No

How was this patch tested?

Manually.

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

No.

[WeichenXu123]

merged to master.

[dongjoon-hyun]

Hi, @WeichenXu123 , @HyukjinKwon , @zhengruifeng .

This sounds like a regression of

• [SPARK-48909][ML][MLLIB] Uses SparkSession over SparkContext when writing metadata #47366

If we cannot get an existing one, I believe we should not create SparkSession here.

Can we recover the existing code?

[HyukjinKwon]

It will not be a regression. This is Spark ML which is DataFrame-based MLlib by definition. Therefore we should always have default session running. Active session is specific to a thread, so it might not exist within the same thread. Alternatively we could use SparkSession.getDefaultSession.

[dongjoon-hyun]

ditto.

[dongjoon-hyun]

ditto.

[dongjoon-hyun]

Initially, the existing PRs assumes that there is no regression because we use the active sessions. AFAIK, this assumption was the same in the dev mailing discussion .

https://lists.apache.org/thread/s24lqtmno0xtoxxz6pk6tyn726bfwp8q

Is this regression inevitable, @HyukjinKwon ?

• If then, could you add a documentation that ML module starts to use SparkSession always instead of SparkContext?
• If that is the module's changed minimum requirement, we don't need to discuss this topic again.

[dongjoon-hyun]

I replied on the existing thread.

• https://lists.apache.org/thread/ks68nys5n5cr129gx35gnx55t4k1x2qb

[HyukjinKwon]

There is no regression. This is Spark ML which is DataFrame-based MLlib. There should be a running Spark session always.

[zhengruifeng]

@dongjoon-hyun
DefaultParamsReader.loadMetadata is only used to load the metadata of ml models, let me take LogisticRegressionModel as an example:

spark/mllib/src/main/scala/org/apache/spark/ml/classification/LogisticRegression.scala

Lines 1322 to 1358 in 2e1a39c

	private class LogisticRegressionModelReader extends MLReader[LogisticRegressionModel] {
	/** Checked against metadata when loading model */
	private val className = classOf[LogisticRegressionModel].getName
	override def load(path: String): LogisticRegressionModel = {
	val metadata = DefaultParamsReader.loadMetadata(path, sc, className)
	val (major, minor) = VersionUtils.majorMinorVersion(metadata.sparkVersion)
	val dataPath = new Path(path, "data").toString
	val data = sparkSession.read.format("parquet").load(dataPath)
	val model = if (major < 2 || (major == 2 && minor == 0)) {
	// 2.0 and before
	val Row(numClasses: Int, numFeatures: Int, intercept: Double, coefficients: Vector) =
	MLUtils.convertVectorColumnsToML(data, "coefficients")
	.select("numClasses", "numFeatures", "intercept", "coefficients")
	.head()
	val coefficientMatrix =
	new DenseMatrix(1, coefficients.size, coefficients.toArray, isTransposed = true)
	val interceptVector = Vectors.dense(intercept)
	new LogisticRegressionModel(metadata.uid, coefficientMatrix,
	interceptVector, numClasses, isMultinomial = false)
	} else {
	// 2.1+
	val Row(numClasses: Int, numFeatures: Int, interceptVector: Vector,
	coefficientMatrix: Matrix, isMultinomial: Boolean) = data
	.select("numClasses", "numFeatures", "interceptVector", "coefficientMatrix",
	"isMultinomial").head()
	new LogisticRegressionModel(metadata.uid, coefficientMatrix, interceptVector,
	numClasses, isMultinomial)
	}
	metadata.getAndSetParams(model)
	model
	}
	}

val metadata = DefaultParamsReader.loadMetadata(path, sc, className)

loads the metadata

val data = sparkSession.read.format("parquet").load(dataPath)

then loads the model coefficients, you can see the sparkSession is already avaiable for model loading.

[zhengruifeng]

I think probably we can change the signature of

def loadMetadata(path: String, sc: SparkContext, expectedClassName: String = ""): Metadata

to

def loadMetadata(path: String, spark: SparkSession, expectedClassName: String = ""): Metadata

to avoid such confusion.

I will have a try

[dongjoon-hyun]

Thank you, @HyukjinKwon and @zhengruifeng . I'm +1 for both to have a clear semantic.

1. Using SparkSession.getDefaultSession instead of *OrCreate.

Alternatively we could use SparkSession.getDefaultSession.

2. Having a clear semantic, def loadMetadata(path: String, spark: SparkSession, expectedClassName: String = ""): Metadata.

[dongjoon-hyun]

For the record and the other reviewers, (2) is implemented and merged to Apache Spark 4.0.0.

• [SPARK-48988][ML] Make DefaultParamsReader/Writer handle metadata with spark session #47467