[SPARK-30659][ML][PYSPARK] LogisticRegression blockify input vectors

[zhengruifeng]

What changes were proposed in this pull request?

1, use blocks instead of vectors
2, use Level-2 BLAS for binary, use Level-3 BLAS for multinomial

Why are the changes needed?

1, less RAM to persist training data; (save ~40%)
2, faster than existing impl; (40% ~ 92%)

Does this PR introduce any user-facing change?

add a new expert param blockSize

How was this patch tested?

updated testsuites

[zhengruifeng]

env: bin/spark-shell --driver-memory=32G

testCode:

importorg.apache.spark.ml.classification._importorg.apache.spark.storage.StorageLevelvardf= spark.read.format("libsvm").load("/data1/Datasets/a9a/a9a").withColumn("label", (col("label")+1)/2)
df.persist(StorageLevel.MEMORY_AND_DISK)
df.count
(0 until 8).foreach{ _ => df = df.union(df) }
df.count
newLogisticRegression().setMaxIter(10).fit(df)
vallr1=newLogisticRegression().setMaxIter(100).setFamily("binomial")
valstart=System.currentTimeMillis; valmodel1= lr1.fit(df); valend=System.currentTimeMillis; end - start
vallr2=newLogisticRegression().setMaxIter(100).setFitIntercept(false).setFamily("binomial")
valstart=System.currentTimeMillis; valmodel2= lr2.fit(df); valend=System.currentTimeMillis; end - start
vallr3=newLogisticRegression().setMaxIter(100).setFamily("multinomial")
valstart=System.currentTimeMillis; valmodel3= lr3.fit(df); valend=System.currentTimeMillis; end - start
vallr4=newLogisticRegression().setMaxIter(100).setFitIntercept(false).setFamily("multinomial")
valstart=System.currentTimeMillis; valmodel4= lr4.fit(df); valend=System.currentTimeMillis; end - start

result:

this PR:
RAM: 1418.9M
DURATION: 136217, 161194, 171625, 177116

Master:
RAM: 2.3G
DURATION: 217035, 218267, 239111, 250163

[zhengruifeng]

Since gradientSumArray is for Matrix of shape CXFPI, and BLAS.gemm requires the output matrix is not transposed. So only if F(numFeature) == FPI(numFeaturesPlusIntercept) and input block is dense, can I use BLAS.gemm to directly update gradientSumArray.
Otherwise, I need to output the result to a temp matrix multinomialLinearGradSumMat, and then add elements to gradientSumArray

[zhengruifeng]

ping @srowen

[srowen]

So, are these lazy just to deal with recreating them after deserialization? they don't seem big, so can they just be non-transient, non-lazy? unless it's a material problem, might be simpler and faster.
Or how much do you need to hold on to scratch vectors like auxiliaryVec vs just locals?

[zhengruifeng]

binaryLinear, binaryIntercept, multinomialLinear, multinomialIntercept are the linear and bias part of coefficients, repectively.

binaryLinearGradSumVec (numFeatures) and multinomialLinearGradSumMat (numClassXnumFeatures) are used to store result of gemv/gemm if fitIntercept==True, since gradientSumArray contains gradient sums of intercepts and can not be used directly in gemv/gemm.

auxiliaryVec (blockSize) and multinomialAuxiliaryMat (blockSizeXnumClasses) are used to store the intermediate multiplication(margins) and multipliers.

they can be used among blocks, and if they are used multi-times in one call we can assign them to local variables.
However I am OK to make them local variables, since I guess they are not the bottleneck.

[srowen]

OK up to your judgment. It'd be simpler to not even make them members, if it's not much difference to performance

[srowen]

Also, does this cause any appreciable slowdown at smaller scale? it's not a big deal if something that's fast is a little slower, to make things that are slow much faster, but just want to get a sense of what you know about the scale implications.

[SparkQA]

Test build #117483 has finished for PR 27374 at commit f753462.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[zhengruifeng]

@srowen
The orignial dataset a9a is not big, its numFeatures=123, numInstances=32,561, after upsampling its numInstances=32,561X256=8,335,616.

I had made other performance tests, it seems that the performance is related to numFeatures and blockSize, and I guess the performance is highly related to: given a array of vectors, to what degree can Level2/3-BLAS be faster than existing java impl or Level-1.

Thanks for reviewing!

[srowen]

Yeah that's the question... Level 1 BLAS often isn't a win. L2/L3 yes. It's probably a win, but just dont' want to take a perf hit on most use cases to help large ones. Even that's arguable.

[zhengruifeng]

Ok, I will test on small datasets.

[SparkQA]

Test build #117501 has finished for PR 27374 at commit 36245b6.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[zhengruifeng]

@srowen I found that on small datasets, the speed up is even more significant.

data: a9a, numFeatures=123, numInstances=32,561

testCode:

importorg.apache.spark.ml.classification._importorg.apache.spark.storage.StorageLevelvaldf= spark.read.format("libsvm").load("/data1/Datasets/a9a/a9a").withColumn("label", (col("label")+1)/2)
df.persist(StorageLevel.MEMORY_AND_DISK)
df.count
vallr4=newLogisticRegression().setMaxIter(100).setFitIntercept(false).setFamily("multinomial")
valstart=System.currentTimeMillis; valmodel4= lr4.fit(df); valend=System.currentTimeMillis; end - start
Seq(64, 256, 1024, 4096, 8192).map { b =>valstart=System.currentTimeMillis; valmodel1=newLogisticRegression().setBlockSize(b).fit(df); valend=System.currentTimeMillis; end - start } // this PRSeq(64, 256, 1024, 4096, 8192).map { b =>valstart=System.currentTimeMillis; valmodel1=newLogisticRegression().fit(df); valend=System.currentTimeMillis; end - start } // Master

result: about 77%~92% faster. I think that is beacuse on big dataset, the communiation overhead has a bigger impact on the whole procedure; while on small datasets like a9a, high-level BLAS dominates the performance.
This PR: List(1630, 1623, 1539, 1559, 1666)
Master: List(2985, 3037, 2957, 2994, 2959)

But the way, I set default value to 1024 base on above result. However, the best blocksize will depend on many factors like whether native-BLAS is used, numFetaures, sparsity, numInstances, etc.

[SparkQA]

Test build #117507 has finished for PR 27374 at commit f0e2e40.

• This patch fails PySpark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[zhengruifeng]

Param BlockSize was just added in #27360, and only used in LinearSVC and LR, so I can safely chang its default value.

[SparkQA]

Test build #117510 has finished for PR 27374 at commit a249fdf.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[srowen]

Looking pretty OK to me

[srowen]

OK up to your judgment. It'd be simpler to not even make them members, if it's not much difference to performance

[srowen]

Would it be faster to fill an array with this value and then make a DenseVector? maybe I'm missing why not

[zhengruifeng]

OK, I will update it.

[srowen]

I'll merge soon to unblock #27389 , but if you have any final thoughts on the above soon, that would be good to check.

[zhengruifeng]

since LeastSquaresAggregator and HuberAggregator in LiR also mark the linear part coefficients/effectiveCoefficientsVector transient and lazy, so I follow this at the begining.
LiR may also benefit from using blocks instead of vectors, I am working on it.

[SparkQA]

Test build #117538 has finished for PR 27374 at commit c49b379.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[srowen]

Merged to master

[WeichenXu123]

@zhengruifeng Could you provide detail benchmark results separately for:

• dense features (all features are dense)
• sparse features (such as 50%, 10%, 1% sparsity)

Thanks!

[mengxr]

+1 on @WeichenXu123 's suggestion and I would suggest temporarily reverting this change before we have a good solution.

@zhengruifeng@srowen This new approach will introduce significant performance regression on sparse datasets with large number of features, e.g., https://www.csie.ntu.edu.tw/~cjlin/libsvmtools/datasets/binary.html#webspam (16,609,143 features). With block size 1024, it requires ~130GB RAM.

[zhengruifeng]

@mengxr@WeichenXu123
I am OK to revert this since it cause regression on high dimensional sparse datasets. Then we may also consider to do revert LinearRegression and LinearSVC, since they three are impled in the same way.

[WeichenXu123]

@zhengruifeng Thanks! Also note your ongoing PR blockify GMM #27473 which do similar thing should also suspend for now.
We have found java-BLAS introduce significant JNI overhead and in some scenario introduce regression.

[zhengruifeng]

@WeichenXu123 Yes, I just mark GMM WIP.
I am going to study this issue, thanks for pointing it out. I am sorry for failing to make a comprehensive test.

[zhengruifeng]

@WeichenXu123@mengxr@srowen
I just made a quick test on webspam:
I draw the first 10,000 sample from webspam_wc_normalized_trigram.svm, and the numFeatures=8,289,919 in the sampled dataset;
It's sparsity (percentage of non-zero values) is about 0.4489%.

This PR will fail dure to OOM in standardization, so I use a patch:

valvec= features match {
casedv: DenseVector=>vari=0while (i < dv.size) {
valstd= featuresStd(i)
if (std !=0) {
dv.values(i) /= std
} else {
dv.values(i) =0.0
}
i +=1
}
dv
casesv: SparseVector=>varj=0while (j < sv.numActives) {
vali= sv.indices(j)
valstd= featuresStd(i)
if (std !=0) {
sv.values(j) /= std
} else {
sv.values(j) =0.0
}
j +=1
}
sv
}

After that, I use following code to test performance:
spark-shell --driver-memory=32G --conf spark.driver.maxResultSize=4g

importorg.apache.spark.ml.classification._importorg.apache.spark.storage.StorageLevelvaldf= spark.read.format("libsvm").load("/data1/Datasets/webspam/webspam_wc_normalized_trigram.svm.10k").withColumn("label", (col("label")+1)/2)
vallr1=newLogisticRegression().setMaxIter(100).setFamily("binomial").setBlockSize(128) // this PRvalstart=System.currentTimeMillis; valmodel1= lr1.fit(df); valend=System.currentTimeMillis; end - start
vallr2=newLogisticRegression().setMaxIter(100).setFamily("binomial").setBlockSize(1024) // this PRvalstart=System.currentTimeMillis; valmodel2= lr2.fit(df); valend=System.currentTimeMillis; end - start
vallr=newLogisticRegression().setMaxIter(100).setFamily("binomial") // 2.4.4valstart=System.currentTimeMillis; valmodel= lr.fit(df); valend=System.currentTimeMillis; end - start

Result:

Impl	This PR(blockSize=128)	This PR(blockSize=1024)	2.4.4
summary.totalIterations	31	31	31
duration	298514	133982	108375
RAM	425	425	396

For this sparse dataset, this PR (with updated standardization) is about 23% slower, and use 7% more RAM.

So I aggre with you to revert this PR and relative PRs LinearSVC, LinearRegression.
Since ALS/MLP extend HasBlockSize depend on LinearSVC, so may it also need to be reverted for now @huaxingao

[srowen]

Ahhh, OK. I didn't think enough about whether sparse vectors would behave significantly differently. Of course, should have been checked. I agree. I'm happy to merge a revert PR or @zhengruifeng you can too.

[huaxingao]

I can revert my PR #27389. But before I revert, I want to check with you folks. My PR only has API changes:

1. add HasBlockSize to ALS so user can specify the blockSize for method blockify
2. make MLP extend HasBlockSize in sharedParams.scala instead of having its own param blockSize

It seems to me it's OK to keep these changes. Of course, there is no HasBlockSize in sharedParams.scala any more after reverting #27360. I guess I can put HasBlockSize in sharedParams.scala instead of reverting my PR?

[mengxr]

I think we can keep the API only change and figure out a way to let the implementation automatically decides whether to blockify+densify for performance.

[zhengruifeng]

OK, I will revert those PRs, and then @huaxingao can add back ALS/MLP extend HasBlockSize as a separate PR.