From f80c8ea08e653e73c66cbad23334116f9248aff4 Mon Sep 17 00:00:00 2001 From: Jinghui Mo Date: Thu, 19 May 2022 16:11:03 -0700 Subject: [PATCH 1/3] Add extensible extractor APIs --- .../feathr/common/AnchorExtractor.scala | 2 +- .../feathr/common/SparkRowExtractor.scala | 23 ++ .../SimpleConfigurableAnchorExtractor.scala | 157 +++----------- .../keyExtractor/MVELSourceKeyExtractor.scala | 9 +- .../offline/config/FeathrConfigLoader.scala | 13 +- .../feathr/offline/config/location/Jdbc.scala | 10 +- .../StreamingFeatureGenerator.scala | 2 +- .../offline/job/FeatureTransformation.scala | 53 ++--- .../dataloader/AvroJsonDataLoader.scala | 3 +- .../jdbc/SnowflakeSqlDataLoader.scala | 2 +- .../DataFrameBasedMVELEvaluator.scala | 71 ------- .../DataFrameBasedRowEvaluator.scala | 200 ++++++++++++++++++ .../DataFrameBasedSqlEvaluator.scala | 5 +- 13 files changed, 296 insertions(+), 254 deletions(-) create mode 100644 src/main/scala/com/linkedin/feathr/common/SparkRowExtractor.scala delete mode 100644 src/main/scala/com/linkedin/feathr/offline/transformation/DataFrameBasedMVELEvaluator.scala create mode 100644 src/main/scala/com/linkedin/feathr/offline/transformation/DataFrameBasedRowEvaluator.scala diff --git a/src/main/scala/com/linkedin/feathr/common/AnchorExtractor.scala b/src/main/scala/com/linkedin/feathr/common/AnchorExtractor.scala index 3476bb6fe..756f2eaf4 100644 --- a/src/main/scala/com/linkedin/feathr/common/AnchorExtractor.scala +++ b/src/main/scala/com/linkedin/feathr/common/AnchorExtractor.scala @@ -4,7 +4,7 @@ package com.linkedin.feathr.common * Provides feature values based on some "raw" data element * @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] diff --git a/src/main/scala/com/linkedin/feathr/common/SparkRowExtractor.scala b/src/main/scala/com/linkedin/feathr/common/SparkRowExtractor.scala new file mode 100644 index 000000000..04e715e8c --- /dev/null +++ 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] +} \ No newline at end of file diff --git a/src/main/scala/com/linkedin/feathr/offline/anchored/anchorExtractor/SimpleConfigurableAnchorExtractor.scala b/src/main/scala/com/linkedin/feathr/offline/anchored/anchorExtractor/SimpleConfigurableAnchorExtractor.scala index 6513b96b6..0cd0b88c1 100644 --- a/src/main/scala/com/linkedin/feathr/offline/anchored/anchorExtractor/SimpleConfigurableAnchorExtractor.scala +++ b/src/main/scala/com/linkedin/feathr/offline/anchored/anchorExtractor/SimpleConfigurableAnchorExtractor.scala @@ -3,20 +3,14 @@ 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 /** @@ -30,7 +24,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 +55,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 +93,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 +121,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 +133,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,7 +175,7 @@ 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 { diff --git a/src/main/scala/com/linkedin/feathr/offline/anchored/keyExtractor/MVELSourceKeyExtractor.scala b/src/main/scala/com/linkedin/feathr/offline/anchored/keyExtractor/MVELSourceKeyExtractor.scala index 0980e3f1a..209ac89e1 100644 --- a/src/main/scala/com/linkedin/feathr/offline/anchored/keyExtractor/MVELSourceKeyExtractor.scala +++ b/src/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 index e08ab6e58..08fe931d6 100644 --- 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, _} 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/config/location/Jdbc.scala b/src/main/scala/com/linkedin/feathr/offline/config/location/Jdbc.scala index 1ebf072fa..b18d5efc5 100644 --- a/src/main/scala/com/linkedin/feathr/offline/config/location/Jdbc.scala +++ b/src/main/scala/com/linkedin/feathr/offline/config/location/Jdbc.scala @@ -5,14 +5,18 @@ import com.fasterxml.jackson.module.caseclass.annotation.CaseClassDeserialize import com.linkedin.feathr.offline.source.dataloader.jdbc.JdbcUtils import com.linkedin.feathr.offline.source.dataloader.jdbc.JdbcUtils.DBTABLE_CONF import org.apache.spark.sql.{DataFrame, SparkSession} -import org.eclipse.jetty.util.StringUtil @CaseClassDeserialize() case class Jdbc(url: String, @JsonAlias(Array("query")) dbtable: String, user: String = "", password: String = "", token: String = "", useToken: Boolean = false, anonymous: Boolean = false) extends InputLocation { + + def isBlankString(str: String): Boolean = { + null == str || "".equals(str) + } + override def loadDf(ss: SparkSession, dataIOParameters: Map[String, String] = Map()): DataFrame = { var reader = ss.read.format("jdbc") .option("url", url) - if (StringUtil.isBlank(dbtable)) { + if (isBlankString(dbtable)) { // Fallback to default table name reader = reader.option("dbtable", ss.conf.get(DBTABLE_CONF)) } else { @@ -30,7 +34,7 @@ case class Jdbc(url: String, @JsonAlias(Array("query")) dbtable: String, user: S .option("encrypt", true) .load } else { - if (StringUtil.isBlank(user) && StringUtil.isBlank(password)) { + if (isBlankString(user) && isBlankString(password)) { if (anonymous) { reader.load() } else { diff --git a/src/main/scala/com/linkedin/feathr/offline/generation/StreamingFeatureGenerator.scala b/src/main/scala/com/linkedin/feathr/offline/generation/StreamingFeatureGenerator.scala index 439deaeb1..a3114a47a 100644 --- 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:_*) diff --git a/src/main/scala/com/linkedin/feathr/offline/job/FeatureTransformation.scala b/src/main/scala/com/linkedin/feathr/offline/job/FeatureTransformation.scala index d6a62c708..23ec97d74 100644 --- a/src/main/scala/com/linkedin/feathr/offline/job/FeatureTransformation.scala +++ b/src/main/scala/com/linkedin/feathr/offline/job/FeatureTransformation.scala @@ -13,7 +13,7 @@ import com.linkedin.feathr.offline.join.DataFrameKeyCombiner import com.linkedin.feathr.offline.source.accessor.{DataSourceAccessor, NonTimeBasedDataSourceAccessor, TimeBasedDataSourceAccessor} import com.linkedin.feathr.offline.swa.SlidingWindowFeatureUtils import com.linkedin.feathr.offline.transformation.FeatureColumnFormat.FeatureColumnFormat -import com.linkedin.feathr.offline.transformation.{DataFrameBasedSqlEvaluator, FeatureColumnFormat, FeatureTypeAccumulator, WindowAggregationEvaluator} +import com.linkedin.feathr.offline.transformation._ import com.linkedin.feathr.offline.util.FeaturizedDatasetUtils.tensorTypeToDataFrameSchema import com.linkedin.feathr.offline.util._ import com.linkedin.feathr.offline.util.datetime.{DateTimeInterval, OfflineDateTimeUtils} @@ -23,7 +23,8 @@ import com.linkedin.feathr.swj.aggregate.AggregationType import com.linkedin.feathr.{common, offline} import org.apache.log4j.Logger import org.apache.spark.rdd.RDD -import org.apache.spark.sql.functions.{col, _} +import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema +import org.apache.spark.sql.functions._ import org.apache.spark.sql.types._ import org.apache.spark.sql.{DataFrame, Row, SparkSession} import org.apache.spark.util.sketch.BloomFilter @@ -74,30 +75,12 @@ private[offline] object FeatureTransformation { // feature name, column prefix type FeatureNameAndColumnPrefix = (String, String) - def getFeatureJoinKey(sourceKeyExtractor: SourceKeyExtractor, withKeyColumnRDD: RDD[_]): Seq[String] = { - if (!sourceKeyExtractor.isInstanceOf[MVELSourceKeyExtractor]) { - throw new FeathrFeatureTransformationException( - ErrorLabel.FEATHR_USER_ERROR, - s"Cannot call apply ${sourceKeyExtractor} on RDD as it's expecting MVELSourceKeyExtractor.") - } - if (withKeyColumnRDD.isEmpty) { - sourceKeyExtractor.getKeyColumnNames(None) - } else { - sourceKeyExtractor.getKeyColumnNames(Some(withKeyColumnRDD.first())) - } - } - - def getFeatureJoinKey(sourceKeyExtractor: SourceKeyExtractor, withKeyColumnDF: DataFrame): Seq[String] = { - // special handle MVEL source key extractor, which has two version of getKeyColumnNames() - sourceKeyExtractor match { - case extractor: MVELSourceKeyExtractor => - if (withKeyColumnDF.head(1).isEmpty) { - extractor.getKeyColumnNames(None) - } else { - extractor.getKeyColumnNames(Some(withKeyColumnDF.first())) - } - case extractor => extractor.getKeyColumnNames() - } + def getFeatureJoinKey(sourceKeyExtractor: SourceKeyExtractor, withKeyColumnDF: DataFrame, featureExtractor: Option[AnyRef] = None): Seq[String] = { + if (withKeyColumnDF.head(1).isEmpty) { + sourceKeyExtractor.getKeyColumnNames(None) + } else { + sourceKeyExtractor.getKeyColumnNames(Some(withKeyColumnDF.first())) + } } // get the feature column prefix which will be appended to all feature columns of the dataframe returned by the transformer @@ -186,24 +169,16 @@ private[offline] object FeatureTransformation { val featureNamePrefixPairs = requestedFeatureRefString.map((_, featureNamePrefix)) // return the feature dataframe, the feature column format and the actual(inferred or user provided) feature types + val featureTypeConfigs = featureAnchorWithSource.featureAnchor.featureTypeConfigs val transformedFeatureData: TransformedResult = featureAnchorWithSource.featureAnchor.extractor match { case transformer: TimeWindowConfigurableAnchorExtractor => WindowAggregationEvaluator.transform(transformer, df, featureNamePrefixPairs, featureAnchorWithSource, inputDateInterval) case transformer: SimpleAnchorExtractorSpark => // transform from avro tensor to FDS format, avro tensor can be shared by online/offline // so that transformation logic can be written only once - DataFrameBasedSqlEvaluator.transform(transformer, df, featureNamePrefixPairs, featureAnchorWithSource) - case transformer: SimpleConfigurableAnchorExtractor => - val featureFormat = FeatureColumnFormat.FDS_TENSOR - // features to calculate, if empty, will calculate all features defined in the extractor - val selectedFeatureNames = if (requestedFeatureRefString.nonEmpty) requestedFeatureRefString else transformer.getProvidedFeatureNames - val FeatureDataFrame(transformedDF, tranformedFeatureTypes) = transformer.transform(transformer, df, selectedFeatureNames) - TransformedResult( - // Re-compute the featureNamePrefixPairs because feature names can be coming from the extractor. - selectedFeatureNames.map((_, featureNamePrefix)), - transformedDF, - selectedFeatureNames.map(c => (c, featureFormat)).toMap, - tranformedFeatureTypes) + DataFrameBasedSqlEvaluator.transform(transformer, df, featureNamePrefixPairs, featureTypeConfigs) + case transformer: AnchorExtractor[_] => + DataFrameBasedRowEvaluator.transform(transformer, df, featureNamePrefixPairs, featureTypeConfigs) case _ => throw new FeathrFeatureTransformationException(ErrorLabel.FEATHR_USER_ERROR, s"cannot find valid Transformer for ${featureAnchorWithSource}") } @@ -329,7 +304,7 @@ private[offline] object FeatureTransformation { } val withKeyColumnDF = keyExtractor.appendKeyColumns(sourceDF) - val outputJoinKeyColumnNames = getFeatureJoinKey(keyExtractor, withKeyColumnDF) + val outputJoinKeyColumnNames = getFeatureJoinKey(keyExtractor, withKeyColumnDF, Some(anchorFeatureGroup.anchorsWithSameSource.head.featureAnchor.extractor)) val filteredFactData = applyBloomFilter((keyExtractor, withKeyColumnDF), bloomFilter) // 1. apply all transformations on the dataframe in sequential order diff --git a/src/main/scala/com/linkedin/feathr/offline/source/dataloader/AvroJsonDataLoader.scala b/src/main/scala/com/linkedin/feathr/offline/source/dataloader/AvroJsonDataLoader.scala index 36e303e47..11d524ffc 100644 --- a/src/main/scala/com/linkedin/feathr/offline/source/dataloader/AvroJsonDataLoader.scala +++ b/src/main/scala/com/linkedin/feathr/offline/source/dataloader/AvroJsonDataLoader.scala @@ -15,6 +15,7 @@ import org.apache.spark.sql.{DataFrame, Row, SparkSession} import java.io.{ByteArrayInputStream, DataInputStream} import java.util import scala.collection.convert.wrapAll._ +import scala.io.Source import scala.reflect.ClassTag import scala.util.Try @@ -103,7 +104,7 @@ private[offline] object AvroJsonDataLoader { val sc = ss.sparkContext require(sc.isLocal) require(path.endsWith(".avro.json")) - val contents = scala.io.Source.fromFile(s"src/test/resources/" + path).getLines().mkString + val contents = Source.fromResource(path).getLines().mkString val jackson = new ObjectMapper(new HoconFactory) val tree = jackson.readTree(contents) val jsonDataArray = tree.get("data") diff --git a/src/main/scala/com/linkedin/feathr/offline/source/dataloader/jdbc/SnowflakeSqlDataLoader.scala b/src/main/scala/com/linkedin/feathr/offline/source/dataloader/jdbc/SnowflakeSqlDataLoader.scala index 312caad7c..e7b89a1c0 100644 --- a/src/main/scala/com/linkedin/feathr/offline/source/dataloader/jdbc/SnowflakeSqlDataLoader.scala +++ b/src/main/scala/com/linkedin/feathr/offline/source/dataloader/jdbc/SnowflakeSqlDataLoader.scala @@ -1,9 +1,9 @@ package com.linkedin.feathr.offline.source.dataloader.jdbc -import org.apache.commons.httpclient.URI import org.apache.http.client.utils.URLEncodedUtils import org.apache.spark.sql.{DataFrame, DataFrameReader, SparkSession} +import java.net.URI import scala.collection.JavaConverters.asScalaBufferConverter import java.nio.charset.Charset diff --git a/src/main/scala/com/linkedin/feathr/offline/transformation/DataFrameBasedMVELEvaluator.scala b/src/main/scala/com/linkedin/feathr/offline/transformation/DataFrameBasedMVELEvaluator.scala deleted file mode 100644 index 92aa16ac1..000000000 --- a/src/main/scala/com/linkedin/feathr/offline/transformation/DataFrameBasedMVELEvaluator.scala +++ /dev/null @@ -1,71 +0,0 @@ -package com.linkedin.feathr.offline.transformation - -import com.linkedin.feathr.common.FeatureTypes -import com.linkedin.feathr.common.exception.{ErrorLabel, FeathrException} -import org.apache.spark.util.AccumulatorV2 - -/** - * Evaluator that transforms features using MVEL based on DataFrame - * It evaluates the extractor class against the input data to get the feature value - * This is used in sequential join's MVEL based expand feature evaluation, and anchored passthrough feature evaluation. - * It is essentially a batch version of getFeatures() for dataframe - */ -private[offline] object DataFrameBasedMVELEvaluator { - - // offline/anchored/anchorExtractor/SimpleConfigurableAnchorExtractor.scala here -} - -// Used to 'accumulate' the feature type for a feature while transforming the feature for each row in the dataset -private[offline] class FeatureTypeAccumulator(var featureType: FeatureTypes) extends AccumulatorV2[FeatureTypes, FeatureTypes] { - - def reset(): Unit = { - featureType = FeatureTypes.UNSPECIFIED - } - - def add(input: FeatureTypes): Unit = { - featureType = updateFeatureType(featureType, input) - } - - def value: FeatureTypes = { - featureType - } - - def isZero: Boolean = { - featureType == FeatureTypes.UNSPECIFIED - } - - def copy(): FeatureTypeAccumulator = { - new FeatureTypeAccumulator(featureType) - } - - def merge(other: AccumulatorV2[FeatureTypes, FeatureTypes]): Unit = { - featureType = updateFeatureType(featureType, other.value) - } - - /** - * Update existing feature type (inferred by previous rows) with the new feature type inferred from a new row - * Rules: - * type + type => type - * Unspecified + type => type - * CATEGORICAL + CATEGORICAL_SET => CATEGORICAL_SET - * DENSE_VECTOR + VECTOR => DENSE_VECTOR - * - * @param existingType existing feature type - * @param currentType new feature type - * @return new feature type - */ - private def updateFeatureType(existingType: FeatureTypes, currentType: FeatureTypes): FeatureTypes = { - if (existingType != currentType) { - (existingType, currentType) match { - case (eType, FeatureTypes.UNSPECIFIED) => eType - case (FeatureTypes.UNSPECIFIED, tType) => tType - case (eType, tType) if (Set(eType, tType).subsetOf(Set(FeatureTypes.CATEGORICAL_SET, FeatureTypes.CATEGORICAL))) => - FeatureTypes.CATEGORICAL_SET - case (eType, tType) if (Set(eType, tType).subsetOf(Set(FeatureTypes.DENSE_VECTOR))) => - FeatureTypes.DENSE_VECTOR - case (eType, tType) => - throw new FeathrException(ErrorLabel.FEATHR_USER_ERROR, s"A feature should have only one feature type, but found ${eType} and ${tType}") - } - } else currentType - } -} diff --git a/src/main/scala/com/linkedin/feathr/offline/transformation/DataFrameBasedRowEvaluator.scala b/src/main/scala/com/linkedin/feathr/offline/transformation/DataFrameBasedRowEvaluator.scala new file mode 100644 index 000000000..186a74361 --- /dev/null +++ b/src/main/scala/com/linkedin/feathr/offline/transformation/DataFrameBasedRowEvaluator.scala @@ -0,0 +1,200 @@ +package com.linkedin.feathr.offline.transformation + +import com.linkedin.feathr.common.exception.{ErrorLabel, FeathrException} +import com.linkedin.feathr.common.tensor.TensorData +import com.linkedin.feathr.common.{AnchorExtractor, FeatureTypeConfig, FeatureTypes, SparkRowExtractor} +import com.linkedin.feathr.offline +import com.linkedin.feathr.offline.FeatureDataFrame +import com.linkedin.feathr.offline.job.{FeatureTransformation, FeatureTypeInferenceContext, TransformedResult} +import com.linkedin.feathr.offline.util.FeaturizedDatasetUtils +import org.apache.spark.rdd.RDD +import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema +import org.apache.spark.sql.types.StructType +import org.apache.spark.sql.{DataFrame, Row, SparkSession} +import org.apache.spark.util.AccumulatorV2 + +/** + * Evaluator that transforms features using MVEL based on DataFrame + * It evaluates the extractor class against the input data to get the feature value + * This is used in sequential join's MVEL based expand feature evaluation, and anchored passthrough feature evaluation. + * It is essentially a batch version of getFeatures() for dataframe + */ +private[offline] object DataFrameBasedRowEvaluator { + + /** + * Apply a transformer to an input Dataframe to produce features + * @param transformer transformer to be applied + * @param inputDf input dataframe + * @param requestedFeatureNameAndPrefix feature name and the prefix map + * @param featureTypeConfigs feature name to feature type definition + * @return Transformed result, e.g. input dataframe with features + */ + def transform(transformer: AnchorExtractor[_], + inputDf: DataFrame, + requestedFeatureNameAndPrefix: Seq[(String, String)], + featureTypeConfigs: Map[String, FeatureTypeConfig]): TransformedResult = { + if (!transformer.isInstanceOf[SparkRowExtractor]) { + throw new FeathrException(ErrorLabel.FEATHR_USER_ERROR, s"${transformer} must extend SparkRowExtractor.") + } + val extractor = transformer.asInstanceOf[SparkRowExtractor] + val requestedFeatureRefString = requestedFeatureNameAndPrefix.map(_._1) + val featureNamePrefix = requestedFeatureNameAndPrefix.head._2 + val featureFormat = FeatureColumnFormat.FDS_TENSOR + // features to calculate, if empty, will calculate all features defined in the extractor + val selectedFeatureNames = if (requestedFeatureRefString.nonEmpty) requestedFeatureRefString else transformer.getProvidedFeatureNames + val FeatureDataFrame(transformedDF, transformedFeatureTypes) = transformToFDSTensor(extractor, inputDf, selectedFeatureNames, featureTypeConfigs) + TransformedResult( + // Re-compute the featureNamePrefixPairs because feature names can be coming from the extractor. + selectedFeatureNames.map((_, featureNamePrefix)), + transformedDF, + selectedFeatureNames.map(c => (c, featureFormat)).toMap, + transformedFeatureTypes) + } + + /** + * 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(rowExtractor: SparkRowExtractor, + inputDF: DataFrame, + featureRefStrs: Seq[String], + featureTypeConfigs: Map[String, FeatureTypeConfig]): FeatureDataFrame = { + val inputSchema = inputDF.schema + val spark = SparkSession.builder().getOrCreate() + val featureTypes = featureTypeConfigs.mapValues(_.getFeatureType) + 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.asInstanceOf[GenericRowWithSchema] + } else { + new GenericRowWithSchema(row.toSeq.toArray, inputSchema) + } + val result = rowExtractor.getFeaturesFromRow(rowWithSchema) + val featureValues = featureRefStrs map { + featureRef => + if (result.contains(featureRef)) { + val featureValue = result(featureRef) + if (featureTypeAccumulators(featureRef).isZero && featureValue != null) { + val rowFeatureType = featureValue.getFeatureType.getBasicType + featureTypeAccumulators(featureRef).add(FeatureTypes.valueOf(rowFeatureType.toString)) + } + val tensorData: TensorData = featureValue.getAsTensorData() + FeaturizedDatasetUtils.tensorToDataFrameRow(tensorData) + } 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 + } +} + +// Used to 'accumulate' the feature type for a feature while transforming the feature for each row in the dataset +private[offline] class FeatureTypeAccumulator(var featureType: FeatureTypes) extends AccumulatorV2[FeatureTypes, FeatureTypes] { + + def reset(): Unit = { + featureType = FeatureTypes.UNSPECIFIED + } + + def add(input: FeatureTypes): Unit = { + featureType = updateFeatureType(featureType, input) + } + + def value: FeatureTypes = { + featureType + } + + def isZero: Boolean = { + featureType == FeatureTypes.UNSPECIFIED + } + + def copy(): FeatureTypeAccumulator = { + new FeatureTypeAccumulator(featureType) + } + + def merge(other: AccumulatorV2[FeatureTypes, FeatureTypes]): Unit = { + featureType = updateFeatureType(featureType, other.value) + } + + /** + * Update existing feature type (inferred by previous rows) with the new feature type inferred from a new row + * Rules: + * type + type => type + * Unspecified + type => type + * CATEGORICAL + CATEGORICAL_SET => CATEGORICAL_SET + * DENSE_VECTOR + VECTOR => DENSE_VECTOR + * + * @param existingType existing feature type + * @param currentType new feature type + * @return new feature type + */ + private def updateFeatureType(existingType: FeatureTypes, currentType: FeatureTypes): FeatureTypes = { + if (existingType != currentType) { + (existingType, currentType) match { + case (eType, FeatureTypes.UNSPECIFIED) => eType + case (FeatureTypes.UNSPECIFIED, tType) => tType + case (eType, tType) if (Set(eType, tType).subsetOf(Set(FeatureTypes.CATEGORICAL_SET, FeatureTypes.CATEGORICAL))) => + FeatureTypes.CATEGORICAL_SET + case (eType, tType) if (Set(eType, tType).subsetOf(Set(FeatureTypes.DENSE_VECTOR))) => + FeatureTypes.DENSE_VECTOR + case (eType, tType) => + throw new FeathrException(ErrorLabel.FEATHR_USER_ERROR, s"A feature should have only one feature type, but found ${eType} and ${tType}") + } + } else currentType + } +} diff --git a/src/main/scala/com/linkedin/feathr/offline/transformation/DataFrameBasedSqlEvaluator.scala b/src/main/scala/com/linkedin/feathr/offline/transformation/DataFrameBasedSqlEvaluator.scala index 780298b45..079f46850 100644 --- a/src/main/scala/com/linkedin/feathr/offline/transformation/DataFrameBasedSqlEvaluator.scala +++ b/src/main/scala/com/linkedin/feathr/offline/transformation/DataFrameBasedSqlEvaluator.scala @@ -22,17 +22,16 @@ private[offline] object DataFrameBasedSqlEvaluator { * @param transformer SimpleAnchorExtractorSpark implementation * @param inputDf input dataframe * @param requestedFeatureNameAndPrefix feature names and prefix pairs - * @param featureAnchorWithSource feature anchor with source that has the transformer + * @param featureTypeConfigs feature name to feature type config * @return TransformedResult or transformed feature data. */ def transform( transformer: SimpleAnchorExtractorSpark, inputDf: DataFrame, requestedFeatureNameAndPrefix: Seq[(String, String)], - featureAnchorWithSource: FeatureAnchorWithSource): TransformedResult = { + featureTypeConfigs: Map[String, FeatureTypeConfig]): TransformedResult = { val requestedFeatureName = requestedFeatureNameAndPrefix.map(_._1) transformer.setInternalParams(SELECTED_FEATURES, s"[${requestedFeatureName.mkString(",")}]") - val featureTypeConfigs = featureAnchorWithSource.featureAnchor.featureTypeConfigs // Get DataFrame schema for tensor based on inferred tensor type. val featureSchemas = requestedFeatureName From 9d395f45352be886e3a012cfcab1b455f0d5c854 Mon Sep 17 00:00:00 2001 From: Jinghui Mo Date: Sun, 29 May 2022 16:55:18 -0700 Subject: [PATCH 2/3] Fix test --- .../feathr/common/AnchorExtractor.scala | 18 +++++++++++++++++- .../SimpleConfigurableAnchorExtractor.scala | 12 +++++++++--- .../offline/config/FeathrConfigLoader.scala | 2 +- .../DataFrameBasedRowEvaluator.scala | 3 ++- 4 files changed, 29 insertions(+), 6 deletions(-) diff --git a/src/main/scala/com/linkedin/feathr/common/AnchorExtractor.scala b/src/main/scala/com/linkedin/feathr/common/AnchorExtractor.scala index 756f2eaf4..2e38e4d04 100644 --- a/src/main/scala/com/linkedin/feathr/common/AnchorExtractor.scala +++ b/src/main/scala/com/linkedin/feathr/common/AnchorExtractor.scala @@ -1,8 +1,11 @@ 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] with SparkRowExtractor { @@ -31,4 +34,17 @@ trait AnchorExtractor[T] extends AnchorExtractorBase[T] with SparkRowExtractor { */ 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/offline/anchored/anchorExtractor/SimpleConfigurableAnchorExtractor.scala b/src/main/scala/com/linkedin/feathr/offline/anchored/anchorExtractor/SimpleConfigurableAnchorExtractor.scala index 0cd0b88c1..a63149020 100644 --- a/src/main/scala/com/linkedin/feathr/offline/anchored/anchorExtractor/SimpleConfigurableAnchorExtractor.scala +++ b/src/main/scala/com/linkedin/feathr/offline/anchored/anchorExtractor/SimpleConfigurableAnchorExtractor.scala @@ -13,6 +13,8 @@ import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema import org.apache.spark.sql.types._ 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) @@ -181,12 +183,16 @@ private[offline] class SimpleConfigurableAnchorExtractor( @JsonProperty("key") k 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/src/main/scala/com/linkedin/feathr/offline/config/FeathrConfigLoader.scala b/src/main/scala/com/linkedin/feathr/offline/config/FeathrConfigLoader.scala index 08fe931d6..167f868b4 100644 --- a/src/main/scala/com/linkedin/feathr/offline/config/FeathrConfigLoader.scala +++ b/src/main/scala/com/linkedin/feathr/offline/config/FeathrConfigLoader.scala @@ -7,7 +7,7 @@ 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.exception.{ErrorLabel, FeathrConfigException, FeathrException} -import com.linkedin.feathr.common.{AnchorExtractor, _} +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} diff --git a/src/main/scala/com/linkedin/feathr/offline/transformation/DataFrameBasedRowEvaluator.scala b/src/main/scala/com/linkedin/feathr/offline/transformation/DataFrameBasedRowEvaluator.scala index 186a74361..0bdb013d1 100644 --- a/src/main/scala/com/linkedin/feathr/offline/transformation/DataFrameBasedRowEvaluator.scala +++ b/src/main/scala/com/linkedin/feathr/offline/transformation/DataFrameBasedRowEvaluator.scala @@ -144,7 +144,8 @@ private[offline] object DataFrameBasedRowEvaluator { } } -// Used to 'accumulate' the feature type for a feature while transforming the feature for each row in the dataset +// Used to infer the feature type for a feature while transforming the feature for each row in the dataset +// See updateFeatureType() for rules being used. private[offline] class FeatureTypeAccumulator(var featureType: FeatureTypes) extends AccumulatorV2[FeatureTypes, FeatureTypes] { def reset(): Unit = { From f9d9d453c3fedd5903e912a0b48fb057b10d77f1 Mon Sep 17 00:00:00 2001 From: Jinghui Mo Date: Sun, 29 May 2022 17:34:24 -0700 Subject: [PATCH 3/3] Revert URI import --- .../linkedin/feathr/offline/config/location/Jdbc.scala | 10 +++------- .../dataloader/jdbc/SnowflakeSqlDataLoader.scala | 2 +- 2 files changed, 4 insertions(+), 8 deletions(-) diff --git a/src/main/scala/com/linkedin/feathr/offline/config/location/Jdbc.scala b/src/main/scala/com/linkedin/feathr/offline/config/location/Jdbc.scala index b18d5efc5..1ebf072fa 100644 --- a/src/main/scala/com/linkedin/feathr/offline/config/location/Jdbc.scala +++ b/src/main/scala/com/linkedin/feathr/offline/config/location/Jdbc.scala @@ -5,18 +5,14 @@ import com.fasterxml.jackson.module.caseclass.annotation.CaseClassDeserialize import com.linkedin.feathr.offline.source.dataloader.jdbc.JdbcUtils import com.linkedin.feathr.offline.source.dataloader.jdbc.JdbcUtils.DBTABLE_CONF import org.apache.spark.sql.{DataFrame, SparkSession} +import org.eclipse.jetty.util.StringUtil @CaseClassDeserialize() case class Jdbc(url: String, @JsonAlias(Array("query")) dbtable: String, user: String = "", password: String = "", token: String = "", useToken: Boolean = false, anonymous: Boolean = false) extends InputLocation { - - def isBlankString(str: String): Boolean = { - null == str || "".equals(str) - } - override def loadDf(ss: SparkSession, dataIOParameters: Map[String, String] = Map()): DataFrame = { var reader = ss.read.format("jdbc") .option("url", url) - if (isBlankString(dbtable)) { + if (StringUtil.isBlank(dbtable)) { // Fallback to default table name reader = reader.option("dbtable", ss.conf.get(DBTABLE_CONF)) } else { @@ -34,7 +30,7 @@ case class Jdbc(url: String, @JsonAlias(Array("query")) dbtable: String, user: S .option("encrypt", true) .load } else { - if (isBlankString(user) && isBlankString(password)) { + if (StringUtil.isBlank(user) && StringUtil.isBlank(password)) { if (anonymous) { reader.load() } else { diff --git a/src/main/scala/com/linkedin/feathr/offline/source/dataloader/jdbc/SnowflakeSqlDataLoader.scala b/src/main/scala/com/linkedin/feathr/offline/source/dataloader/jdbc/SnowflakeSqlDataLoader.scala index e7b89a1c0..312caad7c 100644 --- a/src/main/scala/com/linkedin/feathr/offline/source/dataloader/jdbc/SnowflakeSqlDataLoader.scala +++ b/src/main/scala/com/linkedin/feathr/offline/source/dataloader/jdbc/SnowflakeSqlDataLoader.scala @@ -1,9 +1,9 @@ package com.linkedin.feathr.offline.source.dataloader.jdbc +import org.apache.commons.httpclient.URI import org.apache.http.client.utils.URLEncodedUtils import org.apache.spark.sql.{DataFrame, DataFrameReader, SparkSession} -import java.net.URI import scala.collection.JavaConverters.asScalaBufferConverter import java.nio.charset.Charset