Packages

c

com.linkedin.feathr.sparkcommon

SimpleAnchorExtractorSpark

abstract class SimpleAnchorExtractorSpark extends AnchorExtractorBase[Any]

Spark DataFrame-based simple anchor extractor.

We strongly recommend developer extends this SimpleAnchorExtractorSpark trait (when SQL based syntax is not able to express the transformation logic) to implement customized transformation logic, instead of extends GenericAnchorExtractorSpark, As this trait is more efficient than GenericAnchorExtractorSpark in Feathr. (less expensive join between observation and transformed feature data), as SimpleAnchorExtractorSpark guarantees that all rows/columns in the input DataFrame will be "passed through" into the returned DataFrame which helps to avoid the number of join in Feathr, thanks to the fact that Feathr uses the output dataframe of SimpleAnchorExtractorSpark to apply other feature anchors (as long as the anchors share same source and keyExtractor) to improve performance significantly.

see SQLConfigurableAnchorExtractor, TestxFeatureDataExtractorV2, SimpleConfigurableAnchorExtractor in feathr repo for example

Linear Supertypes
AnchorExtractorBase[Any], Serializable, Serializable, AnyRef, Any
Ordering
  1. Alphabetic
  2. By Inheritance
Inherited
  1. SimpleAnchorExtractorSpark
  2. AnchorExtractorBase
  3. Serializable
  4. Serializable
  5. AnyRef
  6. Any
  1. Hide All
  2. Show All
Visibility
  1. Public
  2. All

Instance Constructors

  1. new SimpleAnchorExtractorSpark()

Abstract Value Members

  1. abstract def getProvidedFeatureNames: Seq[String]

    The names of the features provided by this anchor.

    The names of the features provided by this anchor.

    Definition Classes
    AnchorExtractorBase

Concrete Value Members

  1. final def !=(arg0: Any): Boolean
    Definition Classes
    AnyRef → Any
  2. final def ##(): Int
    Definition Classes
    AnyRef → Any
  3. final def ==(arg0: Any): Boolean
    Definition Classes
    AnyRef → Any
  4. val _params: Option[Config]
    Definition Classes
    AnchorExtractorBase
  5. def aggregateAsColumns(groupedDataFrame: DataFrame): Seq[(String, Column)]

    apply aggregations, e.g.

    apply aggregations, e.g. via sql window function This function will be called after transformAsColumns. If you don't need aggregation, leave this function as is.

    groupedDataFrame

    the source dataframe after 3 processing steps:

    1. SourceKeyExtractor.appendKeyColumns() by key extractor 2. transformAsColumns 3. groupby by feature join keys, i.e., SourceKeyExtractor.getKeyColumnNames()
    returns

    aggregated feature name and aggregation column express

  6. final def asInstanceOf[T0]: T0
    Definition Classes
    Any
  7. def clone(): AnyRef
    Attributes
    protected[lang]
    Definition Classes
    AnyRef
    Annotations
    @throws( ... ) @native()
  8. final def eq(arg0: AnyRef): Boolean
    Definition Classes
    AnyRef
  9. def equals(arg0: Any): Boolean
    Definition Classes
    AnyRef → Any
  10. def finalize(): Unit
    Attributes
    protected[lang]
    Definition Classes
    AnyRef
    Annotations
    @throws( classOf[java.lang.Throwable] )
  11. final def getClass(): Class[_]
    Definition Classes
    AnyRef → Any
    Annotations
    @native()
  12. def getFeatures(dataFrameWithKeyColumns: DataFrame, expectedColSchemas: Map[String, StructType]): Seq[(String, Column)]

    Transform input dataframe to generate feature columns The column names for the features should be the same as the declared feature names, which are the feature names returned by getProvidedFeatureNames().

    Transform input dataframe to generate feature columns The column names for the features should be the same as the declared feature names, which are the feature names returned by getProvidedFeatureNames().

    dataFrameWithKeyColumns

    input dataframe with join key columns appended

    expectedColSchemas

    map from column name to the expected structTypes. This is a contract feathr is setting, so that the user gives back only the expected columns. We do a sanity check to match these column types. We expect the user to also do the same.

    returns

    seq[(prefixed feature column name, feature column)], note that feature columns returned here will be appended to the input dataframe, so the columns returned here must be derived directly from the input dataFrameWithKeyColumns, e.g, you cannot do: val df = dataFrameWithKeyColumns.withColumn("x", ..) and return Seq(("x", df(x))) because the column df(x) is from df and not appendable to the input dataFrameWithKeyColumns anymore, instead, you may do something similar to: val myUDF = udf(...) return Seq(("x", myUDF(dataFrameWithKeyColumns("inputCol")))) This limitation is by design, so that Feathr can append the returned columns to the input dataframe and pass it to other anchors if possible, this will reduce the numbers of join to the observation data and improve performance significantly. SimpleAnchorExtractorSpark should be able to handle most of the use cases, if not, see if you can move some of the transformation logic to the corresponding SourceKeyExtractor and fit the rest transformation logic within this API

  13. final def getSelectedFeatures(): Set[String]

    get the feature list that are currently selected to evaluate, e.g., defined in the feature join/generation config Do not use this function in constructor or store as member variable, as it relies on internalParams which may not be initialized in constructor, ONLY use it in transformAsColumns, aggregateAsColumns and postProcessing

    get the feature list that are currently selected to evaluate, e.g., defined in the feature join/generation config Do not use this function in constructor or store as member variable, as it relies on internalParams which may not be initialized in constructor, ONLY use it in transformAsColumns, aggregateAsColumns and postProcessing

    returns

    selected features of current extractor

  14. def hashCode(): Int
    Definition Classes
    AnyRef → Any
    Annotations
    @native()
  15. def init(params: Config): Unit

    init with parameter map

    init with parameter map

    Definition Classes
    AnchorExtractorBase
  16. final def isInstanceOf[T0]: Boolean
    Definition Classes
    Any
  17. final def ne(arg0: AnyRef): Boolean
    Definition Classes
    AnyRef
  18. final def notify(): Unit
    Definition Classes
    AnyRef
    Annotations
    @native()
  19. final def notifyAll(): Unit
    Definition Classes
    AnyRef
    Annotations
    @native()
  20. def postProcessing(aggregatedDataFrame: DataFrame): Seq[(String, Column)]

    used to apply various processing after aggregation, e.g., topK This function will be called after transformAsColumns and aggregateAsColumns.

    used to apply various processing after aggregation, e.g., topK This function will be called after transformAsColumns and aggregateAsColumns. If you don't need postProcessing, leave this function as is.

    returns

    final output feature name and column expression

  21. final def synchronized[T0](arg0: ⇒ T0): T0
    Definition Classes
    AnyRef
  22. def toString(): String
    Definition Classes
    AnyRef → Any
  23. def transformAsColumns(dataFrameWithKeyColumns: DataFrame): Seq[(String, Column)]

    Transform input dataframe to generate feature columns The column names for the features should be the same as the declared feature names, which are the feature names returned by getProvidedFeatureNames().

    Transform input dataframe to generate feature columns The column names for the features should be the same as the declared feature names, which are the feature names returned by getProvidedFeatureNames().

    dataFrameWithKeyColumns

    input dataframe with join key columns appended

    returns

    seq[(prefixed feature column name, feature column)], note that feature columns returned here will be appended to the input dataframe, so the columns returned here must be derived directly from the input dataFrameWithKeyColumns, e.g, you cannot do: val df = dataFrameWithKeyColumns.withColumn("x", ..) and return Seq(("x", df(x))) because the column df(x) is from df and not appendable to the input dataFrameWithKeyColumns anymore, instead, you may do something similar to: val myUDF = udf(...) return Seq(("x", myUDF(dataFrameWithKeyColumns("inputCol")))) This limitation is by design, so that Feathr can append the returned columns to the input dataframe and pass it to other anchors if possible, this will reduce the numbers of join to the observation data and improve performance significantly. SimpleAnchorExtractorSpark should be able to handle most of the use cases, if not, see if you can move some of the transformation logic to the corresponding SourceKeyExtractor and fit the rest transformation logic within this API

  24. def validateInputSchema(schema: StructType): Unit

    Check the validity of the input DataFrame, raise an exception if the schema is invalid, e.g, does not contain required input columns or has incorrect column types It is the developer's responsibility to validate the input schema's validity

    Check the validity of the input DataFrame, raise an exception if the schema is invalid, e.g, does not contain required input columns or has incorrect column types It is the developer's responsibility to validate the input schema's validity

    schema

    the schema of input dataframe (i.e dataFrameWithKeyColumns in transformAsColumns)

  25. final def wait(): Unit
    Definition Classes
    AnyRef
    Annotations
    @throws( ... )
  26. final def wait(arg0: Long, arg1: Int): Unit
    Definition Classes
    AnyRef
    Annotations
    @throws( ... )
  27. final def wait(arg0: Long): Unit
    Definition Classes
    AnyRef
    Annotations
    @throws( ... ) @native()

Inherited from AnchorExtractorBase[Any]

Inherited from Serializable

Inherited from Serializable

Inherited from AnyRef

Inherited from Any

Ungrouped