feast-dev · feast-ci-bot · Dec 24, 2020 · Dec 9, 2020 · Dec 11, 2020 · Dec 23, 2020
@@ -1,7 +1,7 @@
 #!/usr/bin/env bash
 
 export DISABLE_SERVICE_FIXTURES=1
-export MAVEN_OPTS="-Dmaven.repo.local=/tmp/.m2/repository -DdependencyLocationsEnabled=false"
+export MAVEN_OPTS="-Dmaven.repo.local=/tmp/.m2/repository -DdependencyLocationsEnabled=false -Dmaven.wagon.httpconnectionManager.ttlSeconds=25 -Dmaven.wagon.http.retryHandler.count=3 -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false"
 export MAVEN_CACHE="gs://feast-templocation-kf-feast/.m2.2020-11-17.tar"
 
 infra/scripts/download-maven-cache.sh --archive-uri ${MAVEN_CACHE} --output-dir /tmp

@@ -1,6 +1,6 @@
 #!/usr/bin/env bash
 
-export MAVEN_OPTS="-Dmaven.repo.local=/tmp/.m2/repository -DdependencyLocationsEnabled=false"
+export MAVEN_OPTS="-Dmaven.repo.local=/tmp/.m2/repository -DdependencyLocationsEnabled=false -Dmaven.wagon.httpconnectionManager.ttlSeconds=25 -Dmaven.wagon.http.retryHandler.count=3 -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false"
 export MAVEN_CACHE="gs://feast-templocation-kf-feast/.m2.2020-11-17.tar"
 
 infra/scripts/download-maven-cache.sh --archive-uri ${MAVEN_CACHE} --output-dir /tmp

diff --git a/spark/ingestion/pom.xml b/spark/ingestion/pom.xml
@@ -189,14 +189,14 @@
         <dependency>
             <groupId>com.dimafeng</groupId>
             <artifactId>testcontainers-scala-scalatest_${scala.version}</artifactId>
-            <version>0.38.6</version>
+            <version>0.38.8</version>
             <scope>test</scope>
         </dependency>
 
         <dependency>
             <groupId>com.dimafeng</groupId>
             <artifactId>testcontainers-scala-kafka_${scala.version}</artifactId>
-            <version>0.38.6</version>
+            <version>0.38.8</version>
             <scope>test</scope>
         </dependency>
 

diff --git a/spark/ingestion/src/main/scala/feast/ingestion/BasePipeline.scala b/spark/ingestion/src/main/scala/feast/ingestion/BasePipeline.scala
@@ -55,7 +55,7 @@ trait BasePipeline {
           .set("spark.metrics.conf.*.sink.statsd.port", c.port.toString)
           .set("spark.metrics.conf.*.sink.statsd.period", "30")
           .set("spark.metrics.conf.*.sink.statsd.unit", "seconds")
-          .set("spark.metrics.namespace", jobConfig.mode.toString.toLowerCase)
+          .set("spark.metrics.namespace", s"feast_${jobConfig.mode.toString.toLowerCase}")
           // until proto parser udf will be fixed, we have to use this
           .set("spark.sql.legacy.allowUntypedScalaUDF", "true")
       case None => ()

diff --git a/spark/ingestion/src/main/scala/feast/ingestion/BatchPipeline.scala b/spark/ingestion/src/main/scala/feast/ingestion/BatchPipeline.scala
@@ -16,12 +16,14 @@
  */
 package feast.ingestion
 
+import feast.ingestion.metrics.IngestionPipelineMetrics
 import feast.ingestion.sources.bq.BigQueryReader
 import feast.ingestion.sources.file.FileReader
 import feast.ingestion.validation.{RowValidator, TypeCheck}
 import org.apache.commons.lang.StringUtils
 import org.apache.spark.SparkEnv
-import org.apache.spark.sql.{SaveMode, SparkSession}
+import org.apache.spark.sql.catalyst.encoders.RowEncoder
+import org.apache.spark.sql.{Encoder, Row, SaveMode, SparkSession}
 
 /**
   * Batch Ingestion Flow:
@@ -37,6 +39,7 @@ object BatchPipeline extends BasePipeline {
     val projection =
       inputProjection(config.source, featureTable.features, featureTable.entities)
     val rowValidator = new RowValidator(featureTable, config.source.eventTimestampColumn)
+    val metrics      = new IngestionPipelineMetrics
 
     val input = config.source match {
       case source: BQSource =>
@@ -57,13 +60,16 @@ object BatchPipeline extends BasePipeline {
 
     val projected = input.select(projection: _*).cache()
 
+    implicit def rowEncoder: Encoder[Row] = RowEncoder(projected.schema)
+
     TypeCheck.allTypesMatch(projected.schema, featureTable) match {
       case Some(error) =>
         throw new RuntimeException(s"Dataframe columns don't match expected feature types: $error")
       case _ => ()
     }
 
     val validRows = projected
+      .mapPartitions(metrics.incrementRead)
       .filter(rowValidator.allChecks)
 
     validRows.write
@@ -79,6 +85,7 @@ object BatchPipeline extends BasePipeline {
       case Some(path) =>
         projected
           .filter(!rowValidator.allChecks)
+          .mapPartitions(metrics.incrementDeadLetters)
           .write
           .format("parquet")
           .mode(SaveMode.Append)

diff --git a/spark/ingestion/src/main/scala/feast/ingestion/StreamingPipeline.scala b/spark/ingestion/src/main/scala/feast/ingestion/StreamingPipeline.scala
@@ -19,8 +19,9 @@ package feast.ingestion
 import java.io.File
 import java.util.concurrent.TimeUnit
 
+import feast.ingestion.metrics.IngestionPipelineMetrics
 import feast.ingestion.registry.proto.ProtoRegistryFactory
-import org.apache.spark.sql.{DataFrame, Row, SaveMode, SparkSession}
+import org.apache.spark.sql.{DataFrame, Encoder, Row, SaveMode, SparkSession}
 import org.apache.spark.sql.functions.{expr, struct, udf}
 import feast.ingestion.utils.ProtoReflection
 import feast.ingestion.utils.testing.MemoryStreamingSource
@@ -33,8 +34,8 @@ import org.apache.spark.sql.streaming.StreamingQuery
 import org.apache.spark.sql.avro._
 import org.apache.spark.sql.execution.python.UserDefinedPythonFunction
 import org.apache.spark.sql.execution.streaming.ProcessingTimeTrigger
-import org.apache.spark.sql.expressions.UserDefinedFunction
 import org.apache.spark.sql.types.BooleanType
+import org.apache.spark.sql.catalyst.encoders.RowEncoder
 
 /**
   * Streaming pipeline (currently in micro-batches mode only, since we need to have multiple sinks: redis & deadletters).
@@ -55,8 +56,8 @@ object StreamingPipeline extends BasePipeline with Serializable {
     val featureTable = config.featureTable
     val projection =
       inputProjection(config.source, featureTable.features, featureTable.entities)
-    val rowValidator = new RowValidator(featureTable, config.source.eventTimestampColumn)
-
+    val rowValidator  = new RowValidator(featureTable, config.source.eventTimestampColumn)
+    val metrics       = new IngestionPipelineMetrics
     val validationUDF = createValidationUDF(sparkSession, config)
 
     val input = config.source match {
@@ -103,8 +104,10 @@ object StreamingPipeline extends BasePipeline with Serializable {
           batchDF.withColumn("_isValid", rowValidator.allChecks)
         }
         rowsAfterValidation.persist()
+        implicit def rowEncoder: Encoder[Row] = RowEncoder(rowsAfterValidation.schema)
 
         rowsAfterValidation
+          .mapPartitions(metrics.incrementRead)
           .filter(if (config.doNotIngestInvalidRows) expr("_isValid") else rowValidator.allChecks)
           .write
           .format("feast.ingestion.stores.redis")
@@ -119,6 +122,7 @@ object StreamingPipeline extends BasePipeline with Serializable {
           case Some(path) =>
             rowsAfterValidation
               .filter("!_isValid")
+              .mapPartitions(metrics.incrementDeadLetters)
               .write
               .format("parquet")
               .mode(SaveMode.Append)

diff --git a/spark/ingestion/src/main/scala/feast/ingestion/metrics/IngestionPipelineMetrics.scala b/spark/ingestion/src/main/scala/feast/ingestion/metrics/IngestionPipelineMetrics.scala
@@ -0,0 +1,56 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright 2018-2020 The Feast Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package feast.ingestion.metrics
+
+import org.apache.spark.SparkEnv
+import org.apache.spark.metrics.source.IngestionPipelineMetricSource
+import org.apache.spark.sql.Row
+
+class IngestionPipelineMetrics extends Serializable {
+
+  def incrementDeadLetters(rowIterator: Iterator[Row]): Iterator[Row] = {
+    val materialized = rowIterator.toArray
+    if (metricSource.nonEmpty)
+      metricSource.get.METRIC_DEADLETTER_ROWS_INSERTED.inc(materialized.length)
+
+    materialized.toIterator
+  }
+
+  def incrementRead(rowIterator: Iterator[Row]): Iterator[Row] = {
+    val materialized = rowIterator.toArray
+    if (metricSource.nonEmpty)
+      metricSource.get.METRIC_ROWS_READ_FROM_SOURCE.inc(materialized.length)
+
+    materialized.toIterator
+  }
+
+  private lazy val metricSource: Option[IngestionPipelineMetricSource] = {
+    val metricsSystem = SparkEnv.get.metricsSystem
+    IngestionPipelineMetricsLock.synchronized {
+      if (metricsSystem.getSourcesByName(IngestionPipelineMetricSource.sourceName).isEmpty) {
+        metricsSystem.registerSource(new IngestionPipelineMetricSource)
+      }
+    }
+
+    metricsSystem.getSourcesByName(IngestionPipelineMetricSource.sourceName) match {
+      case Seq(head) => Some(head.asInstanceOf[IngestionPipelineMetricSource])
+      case _         => None
+    }
+  }
+}
+
+private object IngestionPipelineMetricsLock
diff --git a/spark/ingestion/src/main/scala/feast/ingestion/metrics/StatsdReporterWithTags.scala b/spark/ingestion/src/main/scala/feast/ingestion/metrics/StatsdReporterWithTags.scala
@@ -125,8 +125,13 @@ class StatsdReporterWithTags(
   private def reportGauge(name: String, gauge: Gauge[_])(implicit socket: DatagramSocket): Unit =
     formatAny(gauge.getValue).foreach(v => send(fullName(name), v, GAUGE))
 
-  private def reportCounter(name: String, counter: Counter)(implicit socket: DatagramSocket): Unit =
-    send(fullName(name), format(counter.getCount), COUNTER)
+  private def reportCounter(name: String, counter: Counter)(implicit
+      socket: DatagramSocket
+  ): Unit = {
+    val snapshot = counter.getCount
+    send(fullName(name), format(snapshot), COUNTER)
+    counter.dec(snapshot) // reset counter
+  }
 
   private def reportHistogram(name: String, histogram: Histogram)(implicit
       socket: DatagramSocket

diff --git a/spark/ingestion/src/main/scala/org/apache/spark/metrics/source/BaseMetricSource.scala b/spark/ingestion/src/main/scala/org/apache/spark/metrics/source/BaseMetricSource.scala
@@ -0,0 +1,50 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright 2018-2020 The Feast Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.metrics.source
+
+import com.codahale.metrics.MetricRegistry
+import org.apache.spark.SparkEnv
+
+class BaseMetricSource extends Source {
+  override val sourceName: String = ""
+
+  override val metricRegistry: MetricRegistry = new MetricRegistry
+
+  private val sparkConfig = SparkEnv.get.conf
+
+  private val metricLabels = sparkConfig.get("spark.metrics.labels", "")
+
+  private val appId = sparkConfig.get("spark.app.id", "")
+
+  private val executorId = sparkConfig.get("spark.executor.id", "")
+
+  protected def metricWithLabels(name: String) = {
+    if (metricLabels.isEmpty) {
+      name
+    } else {
+      s"$name#$metricLabels,job_id=$appId-$executorId"
+    }
+  }
+
+  protected def counterWithLabels(name: String) = {
+    if (metricLabels.isEmpty) {
+      name
+    } else {
+      s"$name#$metricLabels"
+    }
+  }
+}
diff --git a/...estion/src/main/scala/org/apache/spark/metrics/source/IngestionPipelineMetricSource.scala b/...estion/src/main/scala/org/apache/spark/metrics/source/IngestionPipelineMetricSource.scala
@@ -0,0 +1,31 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright 2018-2020 The Feast Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.metrics.source
+
+class IngestionPipelineMetricSource extends BaseMetricSource {
+  override val sourceName: String = IngestionPipelineMetricSource.sourceName
+
+  val METRIC_DEADLETTER_ROWS_INSERTED =
+    metricRegistry.counter(counterWithLabels("deadletter_count"))
+
+  val METRIC_ROWS_READ_FROM_SOURCE =
+    metricRegistry.counter(counterWithLabels("read_from_source_count"))
+}
+
+object IngestionPipelineMetricSource {
+  val sourceName = "ingestion_pipeline"
+}
diff --git a/spark/ingestion/src/main/scala/org/apache/spark/metrics/source/RedisSinkMetricSource.scala b/spark/ingestion/src/main/scala/org/apache/spark/metrics/source/RedisSinkMetricSource.scala
@@ -16,43 +16,14 @@
  */
 package org.apache.spark.metrics.source
 
-import com.codahale.metrics.MetricRegistry
-import org.apache.spark.SparkEnv
-
-class RedisSinkMetricSource extends Source {
+class RedisSinkMetricSource extends BaseMetricSource {
   override val sourceName: String = RedisSinkMetricSource.sourceName
 
-  override val metricRegistry: MetricRegistry = new MetricRegistry
-
-  private val sparkConfig = SparkEnv.get.conf
-
-  private val metricLabels = sparkConfig.get("spark.metrics.labels", "")
-
-  private val appId = sparkConfig.get("spark.app.id", "")
-
-  private val executorId = sparkConfig.get("spark.executor.id", "")
-
-  private def metricWithLabels(name: String) = {
-    if (metricLabels.isEmpty) {
-      name
-    } else {
-      s"$name#$metricLabels,job_id=$appId-$executorId"
-    }
-  }
-
-  private def counterWithLabels(name: String) = {
-    if (metricLabels.isEmpty) {
-      name
-    } else {
-      s"$name#$metricLabels"
-    }
-  }
-
   val METRIC_TOTAL_ROWS_INSERTED =
-    metricRegistry.counter(counterWithLabels("feast_ingestion_feature_row_ingested_count"))
+    metricRegistry.counter(counterWithLabels("feature_row_ingested_count"))
 
   val METRIC_ROWS_LAG =
-    metricRegistry.histogram(metricWithLabels("feast_ingestion_feature_row_lag_ms"))
+    metricRegistry.histogram(metricWithLabels("feature_row_lag_ms"))
 }
 
 object RedisSinkMetricSource {