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

Asset Hierarchy - Remove single partition deletion flag #859

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Asset Hierarchy - Remove single partition deletion flag #859

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
9ee2fa5
Remove flag
BenChand Nov 2, 2023
339fdb6
Rename flag
BenChand Nov 2, 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
45 changes: 20 additions & 25 deletions src/main/scala/cognite/spark/v1/AssetHierarchyBuilder.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,31 +78,26 @@ class AssetHierarchyBuilder(config: RelationConfig)(val sqlContext: SQLContext)

import CdpConnector.ioRuntime

def delete(data: DataFrame): Unit = {
val partitionedData = if (config.enableSinglePartitionDeleteAssetHierarchy) {
data.repartition(numPartitions = 1)
} else {
data
}

partitionedData.foreachPartition((rows: Iterator[Row]) => {
val deletes = rows.map(r => fromRow[DeleteItemByCogniteId](r))
Stream
.fromIterator[IO](deletes, chunkSize = batchSize)
.chunks
.parEvalMapUnordered(config.parallelismPerPartition) { chunk =>
client.assets
.deleteRecursive(
chunk.toVector.map(_.toCogniteId),
recursive = true,
ignoreUnknownIds = true)
.flatTap(_ => incMetrics(itemsDeleted, chunk.size))
}
.compile
.drain
.unsafeRunSync()
})
}
def delete(data: DataFrame): Unit =
data
.repartition(numPartitions = 1)
.foreachPartition((rows: Iterator[Row]) => {
val deletes = rows.map(r => fromRow[DeleteItemByCogniteId](r))
Stream
.fromIterator[IO](deletes, chunkSize = batchSize)
.chunks
.parEvalMapUnordered(config.parallelismPerPartition) { chunk =>
client.assets
.deleteRecursive(
chunk.toVector.map(_.toCogniteId),
recursive = true,
ignoreUnknownIds = true)
.flatTap(_ => incMetrics(itemsDeleted, chunk.size))
}
.compile
.drain
.unsafeRunSync()
})

def buildFromDf(data: DataFrame): Unit =
// Do not use .collect to run the builder on one of the executors and not on the driver
Expand Down
6 changes: 3 additions & 3 deletions src/main/scala/cognite/spark/v1/DefaultSource.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -406,8 +406,8 @@ object DefaultSource {
val collectMetrics = toBoolean(parameters, "collectMetrics")
val collectTestMetrics = toBoolean(parameters, "collectTestMetrics")

val enableSinglePartitionDeleteAssetHierarchy =
toBoolean(parameters, "enableSinglePartitionDeleteHierarchy", defaultValue = false)
val userSinglePartition =
toBoolean(parameters, "useSinglePartition", defaultValue = false)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
toBoolean(parameters, "useSinglePartition", defaultValue = false)
toBoolean(parameters, "useSinglePartition", defaultValue = true)

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
toBoolean(parameters, "useSinglePartition", defaultValue = false)
toBoolean(parameters, "useSinglePartitionForAssetHierarchyDeletes", defaultValue = true)

not sure about the name...


val saveMode = parseSaveMode(parameters)
val parallelismPerPartition = {
Expand Down Expand Up @@ -459,7 +459,7 @@ object DefaultSource {
subtrees = subtreesOption,
ignoreNullFields = toBoolean(parameters, "ignoreNullFields", defaultValue = true),
rawEnsureParent = toBoolean(parameters, "rawEnsureParent", defaultValue = true),
enableSinglePartitionDeleteAssetHierarchy = enableSinglePartitionDeleteAssetHierarchy
useSinglePartition = userSinglePartition
)
}

Expand Down
2 changes: 1 addition & 1 deletion src/main/scala/cognite/spark/v1/RelationConfig.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ final case class RelationConfig(
subtrees: AssetSubtreeOption,
ignoreNullFields: Boolean,
rawEnsureParent: Boolean,
enableSinglePartitionDeleteAssetHierarchy: Boolean // flag to test whether single partition helps avoid NPE in asset hierarchy builder
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we might want to keep it for potential future investigations if we see weird NPEs again, that be a way to get one of them ready

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah fair enough, I'll rename it to "useSinglePartition" or something

useSinglePartition: Boolean // a flag for helping to test NPEs by moving the data frame to a single partition
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
useSinglePartition: Boolean // a flag for helping to test NPEs by moving the data frame to a single partition
useSinglePartitionForAssetHierarchyDeletion: Boolean // a flag for helping to test NPEs by moving the data frame to a single partition

or something. Maybe invert the name so that default values is false and it'd mean do repartition to 1. Also let's use repartition as default here

) {

/** Desired number of Spark partitions ~= partitions / parallelismPerPartition */
Expand Down
2 changes: 1 addition & 1 deletion src/test/scala/cognite/spark/v1/SparkTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -252,7 +252,7 @@ trait SparkTest {
subtrees = AssetSubtreeOption.Ingest,
ignoreNullFields = true,
rawEnsureParent = false,
enableSinglePartitionDeleteAssetHierarchy = false
useSinglePartition = false
)

private def getCounterSafe(metricName: String): Option[Long] =
Expand Down