Features are not being ingested due to max age overflow (#1209)

Signed-off-by: Oleksii Moskalenko <[email protected]>
feast-dev · Dec 2, 2020 · ad67392 · ad67392
1 parent 3826bd9
commit ad67392
Show file tree

Hide file tree

Showing 5 changed files with 7 additions and 7 deletions.
diff --git a/spark/ingestion/src/main/scala/feast/ingestion/BatchPipeline.scala b/spark/ingestion/src/main/scala/feast/ingestion/BatchPipeline.scala
@@ -72,7 +72,7 @@ object BatchPipeline extends BasePipeline {
       .option("namespace", featureTable.name)
       .option("project_name", featureTable.project)
       .option("timestamp_column", config.source.eventTimestampColumn)
-      .option("max_age", config.featureTable.maxAge.getOrElse(0))
+      .option("max_age", config.featureTable.maxAge.getOrElse(0L))
       .save()
 
     config.deadLetterPath match {

diff --git a/spark/ingestion/src/main/scala/feast/ingestion/IngestionJobConfig.scala b/spark/ingestion/src/main/scala/feast/ingestion/IngestionJobConfig.scala
@@ -95,7 +95,7 @@ case class FeatureTable(
     project: String,
     entities: Seq[Field],
     features: Seq[Field],
-    maxAge: Option[Int] = None
+    maxAge: Option[Long] = None
 )
 
 case class IngestionJobConfig(

diff --git a/spark/ingestion/src/main/scala/feast/ingestion/StreamingPipeline.scala b/spark/ingestion/src/main/scala/feast/ingestion/StreamingPipeline.scala
@@ -87,7 +87,7 @@ object StreamingPipeline extends BasePipeline with Serializable {
           .option("namespace", featureTable.name)
           .option("project_name", featureTable.project)
           .option("timestamp_column", config.source.eventTimestampColumn)
-          .option("max_age", config.featureTable.maxAge.getOrElse(0))
+          .option("max_age", config.featureTable.maxAge.getOrElse(0L))
           .save()
 
         config.deadLetterPath match {

diff --git a/spark/ingestion/src/main/scala/feast/ingestion/stores/redis/SparkRedisConfig.scala b/spark/ingestion/src/main/scala/feast/ingestion/stores/redis/SparkRedisConfig.scala
@@ -24,7 +24,7 @@ case class SparkRedisConfig(
     iteratorGroupingSize: Int = 1000,
     timestampPrefix: String = "_ts",
     repartitionByEntity: Boolean = true,
-    maxAge: Int = 0,
+    maxAge: Long = 0,
     expiryPrefix: String = "_ex"
 )
 
@@ -43,6 +43,6 @@ object SparkRedisConfig {
       entityColumns = parameters.getOrElse(ENTITY_COLUMNS, "").split(","),
       timestampColumn = parameters.getOrElse(TS_COLUMN, "event_timestamp"),
       repartitionByEntity = parameters.getOrElse(ENTITY_REPARTITION, "true") == "true",
-      maxAge = parameters.get(MAX_AGE).map(_.toInt).getOrElse(0)
+      maxAge = parameters.get(MAX_AGE).map(_.toLong).getOrElse(0)
     )
 }
diff --git a/spark/ingestion/src/test/scala/feast/ingestion/BatchPipelineIT.scala b/spark/ingestion/src/test/scala/feast/ingestion/BatchPipelineIT.scala
@@ -131,7 +131,7 @@ class BatchPipelineIT extends SparkSpec with ForAllTestContainer {
       val gen       = rowGenerator(startDate, endDate)
       val rows      = generateDistinctRows(gen, 1000, groupByEntity)
       val tempPath  = storeAsParquet(sparkSession, rows)
-      val maxAge    = 86400 * 2
+      val maxAge    = 86400L * 30
       val configWithMaxAge = config.copy(
         source = FileSource(tempPath, Map.empty, "eventTimestamp"),
         featureTable = config.featureTable.copy(maxAge = Some(maxAge)),
@@ -162,7 +162,7 @@ class BatchPipelineIT extends SparkSpec with ForAllTestContainer {
 
       })
 
-      val increasedMaxAge = 86400 * 3
+      val increasedMaxAge = 86400L * 60
       val configWithSecondFeatureTable = config.copy(
         source = FileSource(tempPath, Map.empty, "eventTimestamp"),
         featureTable = config.featureTable.copy(