feathr-ai · jaymo001 · May 30, 2022 · May 19, 2022 · May 29, 2022 · May 30, 2022
diff --git a/src/main/scala/com/linkedin/feathr/common/AnchorExtractor.scala b/src/main/scala/com/linkedin/feathr/common/AnchorExtractor.scala
@@ -1,10 +1,13 @@
 package com.linkedin.feathr.common
 
+import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
+
 /**
   * Provides feature values based on some "raw" data element
-  * @tparam T raw data type
+ *
+ * @tparam T raw data type
   */
-trait AnchorExtractor[T] extends AnchorExtractorBase[T] {
+trait AnchorExtractor[T] extends AnchorExtractorBase[T] with SparkRowExtractor {
 
   // NOTE: We may want to decouple this from Extractor. It's more a property of the source, or source-reference
   def getKey(datum: T): Seq[String]
@@ -31,4 +34,17 @@ trait AnchorExtractor[T] extends AnchorExtractorBase[T] {
     */
   def getInputType: Class[_]
 
+  /**
+   * Get key from input row
+   * @param datum input row
+   * @return list of feature keys
+   */
+  def getKeyFromRow(datum: GenericRowWithSchema): Seq[String] = getKey(datum.asInstanceOf[T])
+
+  /**
+   * Get the feature value from the row
+   * @param datum input row
+   * @return A map of feature name to feature value
+   */
+  def getFeaturesFromRow(datum: GenericRowWithSchema): Map[String, FeatureValue] = getFeatures(datum.asInstanceOf[T])
 }
diff --git a/src/main/scala/com/linkedin/feathr/common/SparkRowExtractor.scala b/src/main/scala/com/linkedin/feathr/common/SparkRowExtractor.scala
@@ -0,0 +1,23 @@
+package com.linkedin.feathr.common
+
+import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
+
+/**
+ * An extractor trait that provides APIs to transform a Spark GenericRowWithSchema into feature values
+ */
+trait SparkRowExtractor {
+
+  /**
+   * Get key from input row
+   * @param datum input row
+   * @return list of feature keys
+   */
+  def getKeyFromRow(datum: GenericRowWithSchema): Seq[String]
+
+  /**
+   * Get the feature value from the row
+   * @param datum input row
+   * @return A map of feature name to feature value
+   */
+  def getFeaturesFromRow(datum: GenericRowWithSchema): Map[String, FeatureValue]
+}
diff --git a/.../linkedin/feathr/offline/anchored/anchorExtractor/SimpleConfigurableAnchorExtractor.scala b/.../linkedin/feathr/offline/anchored/anchorExtractor/SimpleConfigurableAnchorExtractor.scala
@@ -3,22 +3,18 @@ package com.linkedin.feathr.offline.anchored.anchorExtractor
 import com.fasterxml.jackson.annotation.JsonProperty
 import com.linkedin.feathr.common.exception.{ErrorLabel, FeathrConfigException}
 import com.linkedin.feathr.common.util.CoercionUtils
-import com.linkedin.feathr.common.{AnchorExtractor, FeatureTypeConfig, FeatureTypes, FeatureValue}
+import com.linkedin.feathr.common.{AnchorExtractor, FeatureTypeConfig, FeatureTypes, FeatureValue, SparkRowExtractor}
 import com.linkedin.feathr.offline
-import com.linkedin.feathr.offline.FeatureDataFrame
 import com.linkedin.feathr.offline.config.MVELFeatureDefinition
-import com.linkedin.feathr.offline.job.{FeatureTransformation, FeatureTypeInferenceContext}
 import com.linkedin.feathr.offline.mvel.{MvelContext, MvelUtils}
-import com.linkedin.feathr.offline.transformation.FDSConversionUtils
-import com.linkedin.feathr.offline.util.FeaturizedDatasetUtils.tensorTypeToDataFrameSchema
-import com.linkedin.feathr.offline.util.{FeatureValueTypeValidator, FeaturizedDatasetUtils}
+import com.linkedin.feathr.offline.util.FeatureValueTypeValidator
 import org.apache.log4j.Logger
-import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
 import org.apache.spark.sql.types._
-import org.apache.spark.sql.{DataFrame, Row, SparkSession}
 import org.mvel2.MVEL
 
+import scala.collection.JavaConverters._
+
 /**
  * A general-purpose configurable FeatureAnchor that extract features based on its definitions,
  * the definition includes mvel expression and feature type (optional)
@@ -30,7 +26,7 @@ import org.mvel2.MVEL
  */
 private[offline] class SimpleConfigurableAnchorExtractor( @JsonProperty("key") key: Seq[String],
                                                           @JsonProperty("features") features: Map[String, MVELFeatureDefinition])
-  extends AnchorExtractor[Any] {
+  extends AnchorExtractor[Any] with SparkRowExtractor {
 
   @transient private lazy val log = Logger.getLogger(getClass)
 
@@ -61,6 +57,17 @@ private[offline] class SimpleConfigurableAnchorExtractor( @JsonProperty("key") k
     case (featureRefStr, featureDef) =>
       (featureRefStr, featureDef.featureTypeConfig.getOrElse(new FeatureTypeConfig(FeatureTypes.UNSPECIFIED)))
   }
+
+
+  /**
+   * Get key from input row
+   * @param datum input row
+   * @return list of feature keys
+   */
+  override def getKeyFromRow(datum: GenericRowWithSchema): Seq[String] = {
+    getKey(datum.asInstanceOf[Any])
+  }
+
   override def getKey(datum: Any): Seq[String] = {
     MvelContext.ensureInitialized()
     // be more strict for resolving keys (don't swallow exceptions)
@@ -88,16 +95,25 @@ private[offline] class SimpleConfigurableAnchorExtractor( @JsonProperty("key") k
         (featureRefStr, (MvelUtils.executeExpression(expression, datum, null, featureRefStr), featureType))
     } collect {
       // Apply a partial function only for non-empty feature values, empty feature values will be set to default later
-      case (featureRefStr, (Some(value), _)) =>
-        getFeature(featureRefStr, value)
+      case (featureRefStr, (Some(value), fType)) =>
+        getFeature(featureRefStr, value, fType.getOrElse(FeatureTypes.UNSPECIFIED))
     }
   }
 
+  /**
+   * Extract feature from row
+   * @param row input row
+   *  @return A map of feature name to feature value
+   */
+  override def getFeaturesFromRow(row: GenericRowWithSchema) = {
+    getFeatures(row.asInstanceOf[Any])
+  }
+
   override def getFeatures(datum: Any): Map[String, FeatureValue] = {
     evaluateMVELWithFeatureType(datum, getProvidedFeatureNames.toSet) collect {
       // Apply a partial function only for non-empty feature values, empty feature values will be set to default later
-      case (featureRefStr, (value, _)) =>
-        getFeature(featureRefStr, value)
+      case (featureRefStr, (value, fType)) =>
+        getFeature(featureRefStr, value, fType)
     }
   }
 
@@ -107,21 +123,6 @@ private[offline] class SimpleConfigurableAnchorExtractor( @JsonProperty("key") k
    */
   override def getInputType: Class[_] = classOf[Object]
 
-  /**
-   * Transform and add feature column to input dataframe using mvelExtractor
-
-   * output feature format.
-   * @param mvelExtractor mvel-based anchor extractor
-   * @param inputDF input dataframe
-   * @param selectedFeatureNames features to calculate, if empty, will calculate all features defined in the extractor
-   * @return inputDF with feature columns added in FDS format.
-   */
-  def transform(mvelExtractor: SimpleConfigurableAnchorExtractor, inputDF: DataFrame, selectedFeatureNames: Seq[String]): FeatureDataFrame = {
-    require(selectedFeatureNames.nonEmpty, s"Features to transform cannot be empty.")
-    // Features are extracted to Quince-FDS format by default
-    transformToFDSTensor(mvelExtractor, inputDF, selectedFeatureNames)
-  }
-
   /**
    * get feature type for selected features
    * @param selectedFeatureRefs selected features
@@ -134,103 +135,15 @@ private[offline] class SimpleConfigurableAnchorExtractor( @JsonProperty("key") k
     }
   }
 
-  /**
-   * Transform and add feature column to input dataframe using mvelExtractor, output feature format is FDS Tensor
-   * @param inputDF      input dataframe
-   * @param featureRefStrs features to transform
-   * @return inputDF with feature columns added, and the inferred feature types for features.
-   *         The inferred type is based on looking at the raw feature value returned by MVEL expression. We used to
-   *         infer the feature type and convert them into term-vector in the getFeatures() API. Now in this function,
-   *         we still use the same rules to infer the feature type, but storing them in FDS tensor format instead of
-   *         term-vector.
-   */
-  private def transformToFDSTensor(mvelExtractor: SimpleConfigurableAnchorExtractor, inputDF: DataFrame, featureRefStrs: Seq[String]): FeatureDataFrame = {
-    val inputSchema = inputDF.schema
-    val spark = SparkSession.builder().getOrCreate()
-    val featureTypes = mvelExtractor.getFeatureTypes(featureRefStrs.toSet)
-    val FeatureTypeInferenceContext(featureTypeAccumulators) =
-      FeatureTransformation.getTypeInferenceContext(spark, featureTypes, featureRefStrs)
-    val transformedRdd = inputDF.rdd.map(row => {
-      // in some cases, the input dataframe row here only have Row and does not have schema attached,
-      // while MVEL only works with GenericRowWithSchema, create it manually
-      val rowWithSchema = if (row.isInstanceOf[GenericRowWithSchema]) {
-        row
-      } else {
-        new GenericRowWithSchema(row.toSeq.toArray, inputSchema)
-      }
-      val result = mvelExtractor.evaluateMVELWithFeatureType(rowWithSchema, featureRefStrs.toSet)
-      val featureValues = featureRefStrs map {
-        featureRef =>
-          if (result.contains(featureRef)) {
-            val (featureValue, featureType) = result(featureRef)
-            if (featureTypeAccumulators(featureRef).isZero) {
-              // This is lazy evaluated
-              featureTypeAccumulators(featureRef).add(featureType)
-            }
-            val inferredFeatureType = featureTypeAccumulators(featureRef).value
-            val featureTypeConfig = featureTypeConfigs.getOrElse(featureRef, new FeatureTypeConfig(FeatureTypes.UNSPECIFIED))
-            val tensorType = FeaturizedDatasetUtils.lookupTensorTypeForFeatureRef(featureRef, inferredFeatureType, featureTypeConfig)
-            val schemaType = tensorTypeToDataFrameSchema(tensorType)
-            FDSConversionUtils.rawToFDSRow(featureValue, schemaType)
-          } else null
-      }
-      Row.merge(row, Row.fromSeq(featureValues))
-    })
-    val inferredFeatureTypes = FeatureTransformation.inferFeatureTypes(featureTypeAccumulators, transformedRdd, featureRefStrs)
-    val inferredFeatureTypeConfigs = inferredFeatureTypes.map(featureTypeEntry => featureTypeEntry._1 -> new FeatureTypeConfig(featureTypeEntry._2))
-    // Merge inferred and provided feature type configs
-    val (unspecifiedTypeConfigs, providedTypeConfigs) = featureTypeConfigs.partition(x => x._2.getFeatureType == FeatureTypes.UNSPECIFIED)
-    // providedTypeConfigs should take precedence over inferredFeatureTypeConfigs; inferredFeatureTypeConfigs should
-    // take precedence over unspecifiedTypeConfigs
-    val mergedTypeConfigs = unspecifiedTypeConfigs ++ inferredFeatureTypeConfigs ++ providedTypeConfigs
-    val featureDF: DataFrame = createFDSFeatureDF(inputDF, featureRefStrs, spark, transformedRdd, mergedTypeConfigs)
-    offline.FeatureDataFrame(featureDF, mergedTypeConfigs)
-  }
-
-  /**
-   * Create a FDS feature dataframe from the MVEL transformed RDD.
-   * This functions is used to remove the context fields that have the same names as the feature names,
-   * so that we can support feature with the same name as the context field name, e.g.
-   * features: {
-   *   foo: foo
-   * }
-   * @param inputDF source dataframe
-   * @param featureRefStrs feature ref strings
-   * @param ss spark session
-   * @param transformedRdd transformed RDD[Row] (in FDS format) from inputDF
-   * @param featureTypesConfigs feature types
-   * @return FDS feature dataframe
-   */
-  private def createFDSFeatureDF(
-      inputDF: DataFrame,
-      featureRefStrs: Seq[String],
-      ss: SparkSession,
-      transformedRdd: RDD[Row],
-      featureTypesConfigs: Map[String, FeatureTypeConfig]): DataFrame = {
-    // first add a prefix to the feature column name in the schema
-    val featureColumnNamePrefix = "_feathr_mvel_feature_prefix_"
-    val featureTensorTypeInfo = FeatureTransformation.getFDSSchemaFields(featureRefStrs, featureTypesConfigs,
-      featureColumnNamePrefix)
-
-    val outputSchema = StructType(inputDF.schema.union(StructType(featureTensorTypeInfo)))
-
-    val transformedDF = ss.createDataFrame(transformedRdd, outputSchema)
-    // drop the context column that have the same name as feature names
-    val withoutDupContextFieldDF = transformedDF.drop(featureRefStrs: _*)
-    // remove the prefix we just added, so that we have a dataframe with feature names as their column names
-    val featureDF = featureRefStrs
-      .zip(featureRefStrs)
-      .foldLeft(withoutDupContextFieldDF)((baseDF, namePair) => {
-        baseDF.withColumnRenamed(featureColumnNamePrefix + namePair._1, namePair._2)
-      })
-    featureDF
-  }
-
   /**
    * Helper function to convert a feature data into FeatureValue.
    */
-  private def getFeature(featureRefStr: String, value: Any): (String, FeatureValue) = {
-    val featureTypeConfig = featureTypeConfigs(featureRefStr)
+  private def getFeature(featureRefStr: String, value: Any, featureType: FeatureTypes=FeatureTypes.UNSPECIFIED): (String, FeatureValue) = {
+    val featureTypeConfig = if (featureTypeConfigs(featureRefStr).getFeatureType == FeatureTypes.UNSPECIFIED) {
+      new FeatureTypeConfig(featureType)
+    } else {
+      featureTypeConfigs(featureRefStr)
+    }
     val featureValue = offline.FeatureValue.fromTypeConfig(value, featureTypeConfig)
     FeatureValueTypeValidator.validate(featureValue, featureTypeConfigs(featureRefStr))
     (featureRefStr, featureValue)
@@ -264,18 +177,22 @@ private[offline] class SimpleConfigurableAnchorExtractor( @JsonProperty("key") k
    * @param datum record to evaluate against
    * @return map of (featureRefStr -> (feature value, featureType))
    */
-  private def evaluateMVELWithFeatureType(datum: Any, selectedFeatureRefs: Set[String]): Map[String, (AnyRef, FeatureTypes)] = {
+  def evaluateMVELWithFeatureType(datum: Any, selectedFeatureRefs: Set[String]): Map[String, (AnyRef, FeatureTypes)] = {
     val featureTypeMap = getFeatureTypes(selectedFeatureRefs)
     val result = evaluateMVEL(datum, selectedFeatureRefs)
     val resultWithType = result collect {
       // Apply a partial function only for non-empty feature values, empty feature values will be set to default later
       case (featureRefStr, Some(value)) =>
-
+        val javaValue = if (value.isInstanceOf[scala.collection.Map[_, _]]) {
+          value.asInstanceOf[scala.collection.Map[_, _]].asJava
+        } else {
+          value
+        }
         // If user already provided the feature type, we don't need to coerce/infer it
         val coercedFeatureType = if (featureTypeMap(featureRefStr) == FeatureTypes.UNSPECIFIED) {
-          CoercionUtils.getCoercedFeatureType(value)
+          CoercionUtils.getCoercedFeatureType(javaValue)
         } else featureTypeMap(featureRefStr)
-        (featureRefStr, (value, coercedFeatureType))
+        (featureRefStr, (javaValue, coercedFeatureType))
     }
     resultWithType map {
       case (featureRef, (featureValue, coercedFeatureType)) =>

diff --git a/...main/scala/com/linkedin/feathr/offline/anchored/keyExtractor/MVELSourceKeyExtractor.scala b/...main/scala/com/linkedin/feathr/offline/anchored/keyExtractor/MVELSourceKeyExtractor.scala
@@ -6,6 +6,7 @@ import com.linkedin.feathr.offline.util.AnchorUtils.removeNonAlphaNumChars
 import com.linkedin.feathr.sparkcommon.SourceKeyExtractor
 import org.apache.spark.sql._
 import org.apache.spark.sql.catalyst.encoders.RowEncoder
+import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
 import org.apache.spark.sql.types._
 
 /**
@@ -36,14 +37,14 @@ private[feathr] class MVELSourceKeyExtractor(val anchorExtractorV1: AnchorExtrac
     val encoder = RowEncoder(outputSchema)
     dataFrame
       .map(row => {
-        val keys = anchorExtractorV1.getKey(row)
+        val keys = getKey(row.asInstanceOf[GenericRowWithSchema])
         Row.merge(row, Row.fromSeq(keys))
       })(encoder)
       .toDF()
   }
 
-  def getKey(datum: Any): Seq[String] = {
-    anchorExtractorV1.getKey(datum)
+  def getKey(datum: GenericRowWithSchema): Seq[String] = {
+    anchorExtractorV1.getKeyFromRow(datum)
   }
 
   /**
@@ -54,7 +55,7 @@ private[feathr] class MVELSourceKeyExtractor(val anchorExtractorV1: AnchorExtrac
    */
   override def getKeyColumnNames(datum: Option[Any]): Seq[String] = {
     if (datum.isDefined) {
-      val size = anchorExtractorV1.getKey(datum.get).size
+      val size = getKey(datum.get.asInstanceOf[GenericRowWithSchema]).size
       (1 to size).map(JOIN_KEY_PREFIX + _)
     } else {
       // return empty join key to signal empty dataset

diff --git a/src/main/scala/com/linkedin/feathr/offline/config/FeathrConfigLoader.scala b/src/main/scala/com/linkedin/feathr/offline/config/FeathrConfigLoader.scala
@@ -6,8 +6,8 @@ import com.fasterxml.jackson.databind._
 import com.fasterxml.jackson.databind.module.SimpleModule
 import com.fasterxml.jackson.databind.node._
 import com.jasonclawson.jackson.dataformat.hocon.HoconFactory
-import com.linkedin.feathr.common._
 import com.linkedin.feathr.common.exception.{ErrorLabel, FeathrConfigException, FeathrException}
+import com.linkedin.feathr.common.{AnchorExtractor, AnchorExtractorBase, FeathrJacksonScalaModule, FeatureDerivationFunctionBase, FeatureTypeConfig, FeatureTypes}
 import com.linkedin.feathr.offline
 import com.linkedin.feathr.offline.ErasedEntityTaggedFeature
 import com.linkedin.feathr.offline.anchored.anchorExtractor.{SQLConfigurableAnchorExtractor, SimpleConfigurableAnchorExtractor, TimeWindowConfigurableAnchorExtractor}
@@ -256,14 +256,12 @@ private[offline] class AnchorLoader extends JsonDeserializer[FeatureAnchor] {
 
     anchorExtractor match {
       case extractor: AnchorExtractorBase[_] =>
-        if (node.has("extractor")) {
           val extractorNode = node.get("extractor")
-          if (extractorNode.has("params")) {
+          if (extractorNode != null && extractorNode.get("params") != null) {
             // init the param into the extractor
             val config = ConfigFactory.parseString(extractorNode.get("params").toString)
             extractor.init(config)
           }
-        }
     }
 
     // cast the the extractor class to AnchorExtractor[Any]
@@ -394,13 +392,14 @@ private[offline] class AnchorLoader extends JsonDeserializer[FeatureAnchor] {
               val keyAlias = FeathrConfigLoader.extractStringListOpt(keyNode.get(KEY_ALIAS))
               new SQLSourceKeyExtractor(keys, keyAlias, lateralViewParameters)
             } else {
-              throw new FeathrConfigException(ErrorLabel.FEATHR_USER_ERROR, s"No valid keys found for node ${node}")
+              val anchorExtractor = anchorExtractorBase.asInstanceOf[AnchorExtractor[Any]]
+              new MVELSourceKeyExtractor(anchorExtractor)
             }
           case None =>
-            if (!anchorExtractorBase.isInstanceOf[AnchorExtractor[_]]) {
+            if (!anchorExtractorBase.isInstanceOf[AnchorExtractorBase[_]]) {
               throw new FeathrException(
                 ErrorLabel.FEATHR_USER_ERROR,
-                s"In ${node}, ${anchorExtractorBase} with no key and no keyExtractor must be extends AnchorExtractor")
+                s"In ${node}, ${anchorExtractorBase} with no key and no keyExtractor must be extends AnchorExtractorBase")
             }
             val keyAlias = FeathrConfigLoader.extractStringListOpt(node.get(KEY_ALIAS))
             val anchorExtractor = anchorExtractorBase.asInstanceOf[AnchorExtractor[Any]]

diff --git a/src/main/scala/com/linkedin/feathr/offline/generation/StreamingFeatureGenerator.scala b/src/main/scala/com/linkedin/feathr/offline/generation/StreamingFeatureGenerator.scala
@@ -109,7 +109,7 @@ class StreamingFeatureGenerator {
           val withKeyColumnDF = keyExtractor.appendKeyColumns(rowDF)
           // Apply feature transformation
           val transformedResult = DataFrameBasedSqlEvaluator.transform(anchor.featureAnchor.extractor.asInstanceOf[SimpleAnchorExtractorSpark],
-            withKeyColumnDF, featureNamePrefixPairs, anchor)
+            withKeyColumnDF, featureNamePrefixPairs, anchor.featureAnchor.featureTypeConfigs)
           val outputJoinKeyColumnNames = getFeatureJoinKey(keyExtractor, withKeyColumnDF)
           val selectedColumns = outputJoinKeyColumnNames ++ anchor.selectedFeatures.filter(keyTaggedFeatures.map(_.featureName).contains(_))
           val cleanedDF = transformedResult.df.select(selectedColumns.head, selectedColumns.tail:_*)