Skip to content

Add a retryable mechanism for SWA features when files get deleted #1127

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

Open
rakeshkashyap123 wants to merge 4 commits into
base: main
Choose a base branch
from
Open

Add a retryable mechanism for SWA features when files get deleted #1127

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
6ef4e78
Add a retryable mechanism for SWA features when files get deleted
Mar 29, 2023
717b9ce
Fix few failing tests
Mar 29, 2023
43557c0
fix comment
Mar 29, 2023
20c159e
fix variable naming
Mar 29, 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
Prev Previous commit
Next Next commit
Fix few failing tests
  • Loading branch information
Rakesh Kashyap Hanasoge Padmanabha
Rakesh Kashyap Hanasoge Padmanabha committed Mar 29, 2023
commit 717b9ce0ce5b9198d22183cf66db8fd4afaeaec4
5 changes: 4 additions & 1 deletion feathr-impl/src/main/scala/com/linkedin/feathr/offline/join/DataFrameFeatureJoiner.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -341,13 +341,16 @@ private[offline] class DataFrameFeatureJoiner(logicalPlan: MultiStageJoinPlan, d

// We will retry the SWA features which could not added because of

@jaymo001 jaymo001 Mar 29, 2023

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

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

Fix comment

val retryableErasedEntityTaggedFeatures = requiredWindowAggFeatures.filter(x => retryableFeatureNames.contains(x.getFeatureName))

// Keep only the features which are to be retried.
val updatedWindowAggFeatureStages = windowAggFeatureStages.map(x => (x._1, x._2.intersect(retryableFeatureNames)))
if (retryableFeatureNames.nonEmpty) {
swaJoiner.joinWindowAggFeaturesAsDF(
ss,
featureDataFrame.df,
joinConfig,
keyTagIntsToStrings,
windowAggFeatureStages,
updatedWindowAggFeatureStages,
retryableErasedEntityTaggedFeatures,
bloomFilters,
swaObsTime,
Expand Down
9 changes: 5 additions & 4 deletions ...-impl/src/main/scala/com/linkedin/feathr/offline/swa/SlidingWindowAggregationJoiner.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -263,22 +263,23 @@ private[offline] class SlidingWindowAggregationJoiner(
SlidingWindowFeatureUtils.getFactDataDef(filteredFactData, anchorWithSourceToDFMap.keySet.toSeq, featuresToDelayImmutableMap, selectedFeatures)
}
val origContextObsColumns = labelDataDef.dataSource.columns

val shouldRetryForMissingData = FeathrUtils.getFeathrJobParam(ss.sparkContext.getConf, FeathrUtils.RETRY_ADDING_MISSING_SWA_FEATURES).toBoolean
try {
// THIS IS FOR LOCAL TEST ONLY. It is to induce a spark exception with the root cause of FileNotFoundException.
if (isRetry && LocalFeatureJoinJob.shouldRetryAddingSWAFeatures) throw new SparkException("file not found", new FileNotFoundException())
if (isRetry && FeathrUtils.getFeathrJobParam(ss.sparkContext.getConf, FeathrUtils.LOCAL_RETRY_ADDING_MISSING_SWA_FEATURES).toBoolean)
throw new SparkException("file not found", new FileNotFoundException())
contextDF = if (swaHandler.isDefined) swaHandler.get.join(labelDataDef, factDataDefs.toList) else SlidingWindowJoin.join(labelDataDef, factDataDefs.toList)
} catch {
// Many times the files which are to be loaded gets deleted midway. We will retry all the features at this stage again by reloading the datasets.
case exception: SparkException => if (isRetry && exception.getCause != null && exception.getCause.isInstanceOf[FileNotFoundException]) {
case exception: SparkException => if (isRetry && shouldRetryForMissingData && exception.getCause != null && exception.getCause.isInstanceOf[FileNotFoundException]) {
val unjoinedFeatures = factDataDefs.flatMap(factData => factData.aggFeatures.map(_.name))
retryableSwaFeatures ++= unjoinedFeatures
}
}

val finalJoinedFeatures = joinedFeatures diff retryableSwaFeatures
contextDF = if (shouldFilterNulls && !factDataRowsWithNulls.isEmpty) {
val nullDfWithFeatureCols = joinedFeatures.foldLeft(factDataRowsWithNulls)((s, x) => s.withColumn(x, lit(null)))
val nullDfWithFeatureCols = finalJoinedFeatures.foldLeft(factDataRowsWithNulls)((s, x) => s.withColumn(x, lit(null)))
contextDF.union(nullDfWithFeatureCols)
} else contextDF

Expand Down
7 changes: 6 additions & 1 deletion feathr-impl/src/main/scala/com/linkedin/feathr/offline/util/FeathrUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,9 @@ private[feathr] object FeathrUtils {
val ENABLE_SANITY_CHECK_MODE = "enable.sanity.check.mode"
val SANITY_CHECK_MODE_ROW_COUNT = "sanity.check.row.count"
val FILTER_NULLS = "filter.nulls"
val RETRY_ADDING_MISSING_SWA_FEATURES = "retry.adding.missing.swa.features"
// Retryer to be configured only for local tests
val LOCAL_RETRY_ADDING_MISSING_SWA_FEATURES = "local.retry.adding.missing.swa.features"
val STRING_PARAMETER_DELIMITER = ","

// Used to check if the current dataframe has satisfied the checkpoint frequency
Expand Down Expand Up @@ -86,7 +89,9 @@ private[feathr] object FeathrUtils {
SPARK_JOIN_MIN_PARALLELISM -> (SQLConf.buildConf(getFullConfigKeyName(SPARK_JOIN_MIN_PARALLELISM )).stringConf.createOptional, "10"),
ENABLE_SANITY_CHECK_MODE -> (SQLConf.buildConf(getFullConfigKeyName(ENABLE_SANITY_CHECK_MODE )).stringConf.createOptional, "false"),
SANITY_CHECK_MODE_ROW_COUNT -> (SQLConf.buildConf(getFullConfigKeyName(SANITY_CHECK_MODE_ROW_COUNT )).stringConf.createOptional, "10"),
FILTER_NULLS -> (SQLConf.buildConf(getFullConfigKeyName(FILTER_NULLS )).stringConf.createOptional, "false")
FILTER_NULLS -> (SQLConf.buildConf(getFullConfigKeyName(FILTER_NULLS)).stringConf.createOptional, "false"),
LOCAL_RETRY_ADDING_MISSING_SWA_FEATURES -> (SQLConf.buildConf(getFullConfigKeyName(LOCAL_RETRY_ADDING_MISSING_SWA_FEATURES)).stringConf.createOptional, "false"),
RETRY_ADDING_MISSING_SWA_FEATURES -> (SQLConf.buildConf(getFullConfigKeyName(RETRY_ADDING_MISSING_SWA_FEATURES)).stringConf.createOptional, "true")
)

/**
Expand Down
9 changes: 3 additions & 6 deletions feathr-impl/src/test/scala/com/linkedin/feathr/offline/SlidingWindowAggIntegTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,8 @@
package com.linkedin.feathr.offline

import com.linkedin.feathr.offline.AssertFeatureUtils.{rowApproxEquals, validateRows}
import com.linkedin.feathr.offline.job.LocalFeatureJoinJob
import com.linkedin.feathr.offline.util.FeathrUtils
import com.linkedin.feathr.offline.util.FeathrUtils.{FILTER_NULLS, SKIP_MISSING_FEATURE, setFeathrJobParam}
import com.linkedin.feathr.offline.util.FeathrUtils.{FILTER_NULLS, LOCAL_RETRY_ADDING_MISSING_SWA_FEATURES, SKIP_MISSING_FEATURE, setFeathrJobParam}
import org.apache.spark.sql.Row
import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
import org.apache.spark.sql.types.{LongType, StructField, StructType}
Expand Down Expand Up @@ -335,7 +334,7 @@ class SlidingWindowAggIntegTest extends FeathrIntegTest {
*/
@Test
def testLocalAnchorSWAWithDenseVectorWithRetry(): Unit = {
LocalFeatureJoinJob.shouldRetryAddingSWAFeatures = true
setFeathrJobParam(LOCAL_RETRY_ADDING_MISSING_SWA_FEATURES, "true")
val res = runLocalFeatureJoinForTest(
"""
| settings: {
Expand Down Expand Up @@ -400,7 +399,7 @@ class SlidingWindowAggIntegTest extends FeathrIntegTest {
assertEquals(featureList(0).getAs[Row]("aEmbedding"), mutable.WrappedArray.make(Array(5.5f, 5.8f)))
assertEquals(featureList(0).getAs[Row]("memberEmbeddingAutoTZ"),
TestUtils.build1dSparseTensorFDSRow(Array(0, 1), Array(5.5f, 5.8f)))
LocalFeatureJoinJob.shouldRetryAddingSWAFeatures = false
setFeathrJobParam(LOCAL_RETRY_ADDING_MISSING_SWA_FEATURES, "true")
}

/**
Expand All @@ -410,7 +409,6 @@ class SlidingWindowAggIntegTest extends FeathrIntegTest {
*/
@Test
def testLocalAnchorSWAWithDenseVector(): Unit = {
LocalFeatureJoinJob.shouldRetry = true
val res = runLocalFeatureJoinForTest(
"""
| settings: {
Expand Down Expand Up @@ -475,7 +473,6 @@ class SlidingWindowAggIntegTest extends FeathrIntegTest {
assertEquals(featureList(0).getAs[Row]("aEmbedding"), mutable.WrappedArray.make(Array(5.5f, 5.8f)))
assertEquals(featureList(0).getAs[Row]("memberEmbeddingAutoTZ"),
TestUtils.build1dSparseTensorFDSRow(Array(0, 1), Array(5.5f, 5.8f)))
LocalFeatureJoinJob.shouldRetry = false
}

/**
Expand Down