Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[ML-273] adjust data cache and make more efficient #281

Merged
merged 19 commits into from
Apr 28, 2023
Merged

[ML-273] adjust data cache and make more efficient #281

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
19 commits
Select commit Hold shift + click to select a range
0dc99f1
Migrate KMeans daal to DPC++ (#209)
minmingzhu Jul 14, 2022
bf24b56
Migrate pca daal to DPC++ (#223)
minmingzhu Aug 9, 2022
ac9455a
[ML-226] Migrate correlation daal to DPC++ (#215)
minmingzhu Aug 31, 2022
abdeff5
[ML-227] Migrate moments daal to DPC++ (#229)
minmingzhu Sep 13, 2022
e456a56
change select_device to assign gpus
minmingzhu Nov 21, 2022
2190ddb
update code style
minmingzhu Nov 21, 2022
c28bc78
update GPU.cpp
minmingzhu Dec 5, 2022
a60f863
update
minmingzhu Apr 10, 2023
cdcd75c
update GPU.cpp
minmingzhu Apr 10, 2023
4490825
update
minmingzhu Apr 14, 2023
1592d11
update
minmingzhu Apr 17, 2023
c9cad8a
update
minmingzhu Apr 17, 2023
d8bca43
update
minmingzhu Apr 17, 2023
9d7adac
update
minmingzhu Apr 17, 2023
05bddf1
data cache
minmingzhu Apr 17, 2023
17d29ff
data cache
minmingzhu Apr 17, 2023
40a44c8
data cache
minmingzhu Apr 17, 2023
e7e0420
update scala style
minmingzhu Apr 17, 2023
f2760b0
update
minmingzhu Apr 25, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion mllib-dal/src/main/scala/com/intel/oap/mllib/OneDAL.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -675,7 +675,6 @@ object OneDAL {
.coalesce(executorNum,
partitionCoalescer = Some(new ExecutorInProcessCoalescePartitioner()))
.setName("coalescedRdd")
.cache()

// convert RDD to HomogenTable
val coalescedTables = coalescedRdd.mapPartitionsWithIndex { (index: Int, it: Iterator[Vector]) =>
Expand Down
10 changes: 9 additions & 1 deletion mllib-dal/src/main/scala/org/apache/spark/ml/feature/spark321/PCA.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ import org.apache.spark.mllib.feature
import org.apache.spark.mllib.feature.{PCAModel => OldPCAModel}
import org.apache.spark.mllib.linalg.{Vectors => OldVectors}
import org.apache.spark.sql._
import org.apache.spark.storage.StorageLevel

/**
* PCA trains a model to project vectors to a lower dimensional space of the top `PCA!.k`
Expand All @@ -50,11 +51,15 @@ class PCA @Since("1.5.0") (
@Since("2.0.0")
override def fit(dataset: Dataset[_]): PCAModel = {
transformSchema(dataset.schema, logging = true)
val handlePersistence = (dataset.storageLevel == StorageLevel.NONE)
val input = dataset.select($(inputCol)).rdd
val inputVectors = input.map {
case Row(v: Vector) => v
}

if (handlePersistence) {
inputVectors.persist(StorageLevel.MEMORY_AND_DISK)
inputVectors.count()
}
val numFeatures = inputVectors.first().size
require($(k) <= numFeatures,
s"source vector size $numFeatures must be no less than k=$k")
Expand All @@ -68,6 +73,9 @@ class PCA @Since("1.5.0") (
val executor_cores = Utils.sparkExecutorCores()
val pca = new PCADALImpl(k = $(k), executor_num, executor_cores)
val pcaDALModel = pca.train(inputVectors)
if (handlePersistence) {
inputVectors.unpersist()
}
new OldPCAModel(pcaDALModel.k, pcaDALModel.pc, pcaDALModel.explainedVariance)
} else {
val inputOldVectors = inputVectors.map {
Expand Down
9 changes: 5 additions & 4 deletions mllib-dal/src/main/scala/org/apache/spark/ml/stat/spark321/Correlation.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,12 +74,13 @@ class Correlation extends CorrelationShim {
dataset.sparkSession.sparkContext)
if (Utils.isOAPEnabled() && isPlatformSupported && method == "pearson") {
val handlePersistence = (dataset.storageLevel == StorageLevel.NONE)
if (handlePersistence) {
dataset.persist(StorageLevel.MEMORY_AND_DISK)
}
val rdd = dataset.select(column).rdd.map {
case Row(v: Vector) => v
}
if (handlePersistence) {
rdd.persist(StorageLevel.MEMORY_AND_DISK)
rdd.count()
}
val executor_num = Utils.sparkExecutorNum(dataset.sparkSession.sparkContext)
val executor_cores = Utils.sparkExecutorCores()
val matrix = new CorrelationDALImpl(executor_num, executor_cores)
Expand All @@ -88,7 +89,7 @@ class Correlation extends CorrelationShim {
val schema = StructType(Array(StructField(name, SQLDataTypes.MatrixType, nullable = false)))
val dataframe = dataset.sparkSession.createDataFrame(Seq(Row(matrix)).asJava, schema)
if (handlePersistence) {
dataset.unpersist()
rdd.unpersist()
}
dataframe
} else {
Expand Down
9 changes: 5 additions & 4 deletions mllib-dal/src/main/scala/org/apache/spark/mllib/stat/spark321/Statistics.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,18 +56,19 @@ class Statistics extends SummarizerShim {
X.sparkContext)
if (Utils.isOAPEnabled() && isPlatformSupported) {
val handlePersistence = (X.getStorageLevel == StorageLevel.NONE)
if (handlePersistence) {
X.persist(StorageLevel.MEMORY_AND_DISK)
}
val rdd = X.map {
v => v.asML
}
if (handlePersistence) {
rdd.persist(StorageLevel.MEMORY_AND_DISK)
rdd.count()
}
val executor_num = Utils.sparkExecutorNum(X.sparkContext)
val executor_cores = Utils.sparkExecutorCores()
val summary = new SummarizerDALImpl(executor_num, executor_cores)
.computeSummarizerMatrix(rdd)
if (handlePersistence) {
X.unpersist()
rdd.unpersist()
}
summary
} else {
Expand Down