Stream Feature View Usage¶

In [0]:

import tecton
import pandas
from datetime import datetime

ws = tecton.get_workspace("prod")

import tecton import pandas from datetime import datetime ws = tecton.get_workspace("prod")

Load a Stream Feature¶

In [0]:

fv = ws.get_feature_view("last_transaction_amount_sql")
fv.summary()

fv = ws.get_feature_view("last_transaction_amount_sql") fv.summary()

Out[3]:

Name	last_transaction_amount_sql
Workspace	prod
Description	Last user transaction amount (stream calculated)
Created At	2022-06-02 23:52:03 UTC
Owner
Last Modified By	david@tecton.ai
Family
Source Filename	fraud/features/stream_features/last_transaction_amount_sql.py
Tags	{}
Type	StreamFeatureView
URL	https://staging.tecton.ai/app/repo/prod/features/last_transaction_amount_sql
Entities	fraud_user
Features	amt
Feature Services
Transformation	last_transaction_amount_sql
Timestamp Key	timestamp
Online Materialization	Enabled
Offline Materialization	Enabled
Feature Start Time	2022-05-01 00:00:00 UTC
Online Join Keys	user_id
Offline Join Keys	user_id
Serving TTL	30 days
Schedule Interval	1 days
Online Serving Status	Running
Online Serving Freshness	18s
Materialization Status	[2022-04-30 00:00:00 UTC, 2022-06-07 00:00:00 UTC] -> Ok

Start a Streaming Job for FeatureView Transformation Results¶

In [0]:

x = fv.run_stream(output_temp_table="output_temp_table")

x = fv.run_stream(output_temp_table="output_temp_table")

In [0]:

# Query the result from the streaming output table.
display(spark.sql("SELECT * FROM output_temp_table ORDER BY timestamp DESC LIMIT 5"))

# Query the result from the streaming output table. display(spark.sql("SELECT * FROM output_temp_table ORDER BY timestamp DESC LIMIT 5"))

timestamp	user_id	amt
2022-06-07T18:31:24.393+0000	user_469998441571	54.46
2022-06-07T18:31:21.720+0000	user_460877961787	73.02
2022-06-07T18:31:20.171+0000	user_650387977076	46.05
2022-06-07T18:31:17.291+0000	user_699668125818	59.24
2022-06-07T18:31:15.702+0000	user_394495759023	11.38

Get a Range of Feature Values from Offline Feature Store¶

In [0]:

fv.get_historical_features(start_time=datetime(2022, 5, 1), end_time=datetime(2022, 5, 2)).to_pandas().head()

fv.get_historical_features(start_time=datetime(2022, 5, 1), end_time=datetime(2022, 5, 2)).to_pandas().head()

Out[6]:

	user_id	amt	timestamp
0	user_930691958107	124.70	2022-05-01 19:18:51.218408
1	user_131340471060	102.60	2022-05-01 19:18:53.611035
2	user_687958452057	15.22	2022-05-01 19:18:56.445318
3	user_884240387242	3.40	2022-05-01 19:18:59.420374
4	user_600003278485	62.30	2022-05-01 19:19:02.065633

Read the Latest Features from Online Feature Store¶

In [0]:

fv.get_online_features({"user_id": "user_930691958107"}).to_dict()

fv.get_online_features({"user_id": "user_930691958107"}).to_dict()

Out[7]: {'amt': 180.6}

Read Historical Features from Offline Feature Store with Time-Travel¶

In [0]:

spine_df = pandas.DataFrame({
  'user_id': ['user_930691958107', 'user_131340471060'],
  'timestamp': [datetime(2022, 5, 1, 19), datetime(2022, 5, 6, 10)]
})
display(spine_df)

spine_df = pandas.DataFrame({ 'user_id': ['user_930691958107', 'user_131340471060'], 'timestamp': [datetime(2022, 5, 1, 19), datetime(2022, 5, 6, 10)] }) display(spine_df)

user_id	timestamp
user_930691958107	2022-05-01T19:00:00.000+0000
user_131340471060	2022-05-06T10:00:00.000+0000

In [0]:

features_df = fv.get_historical_features(spine_df).to_pandas()
display(features_df)

features_df = fv.get_historical_features(spine_df).to_pandas() display(features_df)

INFO - 06/07/2022 06:30:43 PM - Query Builder - last_transaction_amount_sql: (FV 1/1) Start Build [1/3] INFO - 06/07/2022 06:30:43 PM - Query Builder - last_transaction_amount_sql: Building Tile DataFrame [2/3] INFO - 06/07/2022 06:30:44 PM - Query Builder - last_transaction_amount_sql: Generating Join [3/3]

user_id	timestamp	last_transaction_amount_sql__amt
user_930691958107	2022-05-01T19:00:00.000+0000	31.67
user_131340471060	2022-05-06T10:00:00.000+0000	58.68