Skip to content

Add feature transformation support to FeatureView #4277

Closed

Description

@franciscojavierarceo

Is your feature request related to a problem? Please describe.
FeatureViews should support transformations as well.

Describe the solution you'd like

@batch_feature_view(
    sources=[credit_data_batch],
    entities=[user],
    mode="python",
    batch_schedule=timedelta(days=1),
    schema=[Field("user_id", String), Field("timestamp", Timestamp), Field("current_balance", Float64)],
)
def user_last_balance(transactions):
    return transactions[["user_id", "timestamp", "current_balance"]]

Describe alternatives you've considered
N/A

Additional context
Should behave similar to ODFV and Stream Feature Views and support Python and Pandas.

Activity

  1. tokoko commented on Jul 8, 2024

    @tokoko
    Collaborator

    @franciscojavierarceo Just to make sure we're on the same page, do you think we should have a different BatchFeatureView class for this just like odfvs and stream feature views, right?

    Also, I think we should probably implement it w/o scheduled materialization at first, because it's not really obvious to me which feast service could do the scheduling for this (flight server seems most logical or maybe we need to add another service altogether).

  2. franciscojavierarceo commented on Jul 8, 2024

    @franciscojavierarceo
    MemberAuthor

    I am starting to think having the decorator of feature view type is weird.

    Instead we could just do something like:

    @transform(
        sources=[credit_data_batch],
        entities=[user],
        mode="python",
        batch_schedule=timedelta(days=1),
        schema=[Field("user_id", String), Field("timestamp", Timestamp), Field("current_balance", Float64)],
    )
    def user_last_balance(transactions):
        return some_computation(transactions[["user_id", "timestamp", "current_balance"]])

    And maybe in the decorator we have a way to map to when the computation is happening?

    My view is that we need to provide clarity about when transformations happen (on demand, during a write, or in a stream--i.e., before a write) and I'm not sure the current way does this as obviously. Maybe it does.

  3. franciscojavierarceo commented on Aug 12, 2024

    @franciscojavierarceo
    MemberAuthor

    This will be solved #4376

  4. franciscojavierarceo commented on Aug 12, 2024

    @franciscojavierarceo
    MemberAuthor

    Actually this is more general and does not handle the batch transformation use case so keeping this open. The ODFV writes do help for a subset of these items.

  5. Vishnu-Rangiah commented on Oct 21, 2024

    @Vishnu-Rangiah

    Chaining transformation into a dag which can be reused across FVs would be nice to have. This functionality would reduce code duplication across the feature repo. Here is an example adapted from tecton's transformation API: https://docs.tecton.ai/docs/defining-features/feature-views/transformations#a-feature-view-that-calls-a-pyspark-transformation-passing-two-pyspark-transformation-outputs

    @transformation(mode="big_query")
    def last_balance_time(transactions, max_user_transaction):
        return f"""SELECT t.user_id, t.current_balance, last_t.last_transaction_date as timestamp
                          FROM {transactions} t
                          INNER JOIN {max_user_transaction} last_t
                          ON t.user_id = last_t.user_id AND t.timestamp = last_t.last_transaction_date;"""
    
    @transformation(mode="big_query")
    def user_last_transaction_time(transactions):
        return f"""SELECT user_id, MAX(timestamp) AS last_transaction_date
                          FROM {transactions}
                          GROUP BY user_id"""
    
    @feature_view(
        sources=[credit_data_batch],
        entities=[user],
        mode="pipeline", # creates a DAG from re-useable transformation functions
        batch_schedule=timedelta(days=1),
        schema=[Field("user_id", String), Field("timestamp", Timestamp), Field("current_balance", Float64)],
    )
    def user_last_balance(transactions):
    
        user_last_transaction_time = user_last_transaction_time(transactions)
        return last_balance_time(transactions, user_last_transaction_time)
    
  6. franciscojavierarceo commented on Mar 4, 2025

    @franciscojavierarceo
    MemberAuthor

    Will be solved by #4584

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions