Sdk dlt example

In [1]:

            
                Copied!
                
import time
import os 

import featurebyte as fb
import time
import os 

import featurebyte as fb

10:51:42 | INFO     | Using configuration file at: /Users/viktor/.featurebyte/config.yaml
10:51:42 | INFO     | Active profile: local (http://127.0.0.1:8088)
10:51:42 | INFO     | SDK version: 0.6.0
10:51:42 | INFO     | No catalog activated.
10:51:42 | INFO     | 0 feature list, 0 feature deployed

Connect to databricks data warehouse¶

In [2]:

            
                Copied!
                
                    
                    
                
                

        
ds = fb.FeatureStore.get_or_create(
    name="databricks",
    source_type=fb.SourceType.DATABRICKS,
    details=fb.DatabricksDetails(
        host="<host>",
        http_path="<http_path>",
        featurebyte_catalog="hive_metastore",
        featurebyte_schema="grocery",        
        storage_type=fb.StorageType.S3,
        storage_url=f"<storage_url>",
        storage_spark_url=f"<storage_spark_url>",
    ),
    database_credential=fb.AccessTokenCredential(
        access_token="<databricks_access_token>"
    ),
    storage_credential=fb.S3StorageCredential(
        s3_access_key_id="<s3_access_key_id>",
        s3_secret_access_key="<s3_secret_access_key>",
    )
).get_data_source()
ds = fb.FeatureStore.get_or_create(
    name="databricks",
    source_type=fb.SourceType.DATABRICKS,
    details=fb.DatabricksDetails(
        host="",
        http_path="",
        featurebyte_catalog="hive_metastore",
        featurebyte_schema="grocery",        
        storage_type=fb.StorageType.S3,
        storage_url=f"",
        storage_spark_url=f"",
    ),
    database_credential=fb.AccessTokenCredential(
        access_token=""
    ),
    storage_credential=fb.S3StorageCredential(
        s3_access_key_id="",
        s3_secret_access_key="",
    )
).get_data_source()

Create catalog¶

In [3]:

            
                Copied!
                
catalog_name = "DLT Example Catalog"
fb.Catalog.create(catalog_name, feature_store_name="databricks")
catalog = fb.Catalog.activate(catalog_name)
catalog_name = "DLT Example Catalog"
fb.Catalog.create(catalog_name, feature_store_name="databricks")
catalog = fb.Catalog.activate(catalog_name)

10:51:44 | INFO     | Catalog activated: DLT Example Catalog

Create tables¶

In [4]:

            
                Copied!
                
                    
                    
                
                

        
%%capture 

customer_table = ds.get_source_table(
    database_name="hive_metastore", schema_name="grocery", table_name="grocerycustomer"
).create_scd_table(
    name="grocerycustomer",
    surrogate_key_column='RowID',
    natural_key_column="GroceryCustomerGuid",
    effective_timestamp_column="ValidFrom",
    current_flag_column="CurrentRecord",
    record_creation_timestamp_column="record_available_at",
)
invoice_table = ds.get_source_table(
    database_name="hive_metastore", schema_name="grocery", table_name="groceryinvoice"
).create_event_table(
    name="groceryinvoice",
    event_id_column="GroceryInvoiceGuid",
    event_timestamp_column="Timestamp",
    event_timestamp_timezone_offset_column="tz_offset",
    record_creation_timestamp_column="record_available_at",
)
invoice_table.initialize_default_feature_job_setting()
items_table = ds.get_source_table(
    database_name="hive_metastore", schema_name="grocery", table_name="invoiceitems"
).create_item_table(
    name="invoiceitems",
    event_id_column="GroceryInvoiceGuid",
    item_id_column="GroceryInvoiceItemGuid",
    event_table_name="groceryinvoice",
    record_creation_timestamp_column="record_available_at",
)
product_table = ds.get_source_table(
    database_name="hive_metastore", schema_name="grocery", table_name="groceryproduct_embedding"
).create_dimension_table(name="groceryproduct", dimension_id_column="GroceryProductGuid")
%%capture 

customer_table = ds.get_source_table(
    database_name="hive_metastore", schema_name="grocery", table_name="grocerycustomer"
).create_scd_table(
    name="grocerycustomer",
    surrogate_key_column='RowID',
    natural_key_column="GroceryCustomerGuid",
    effective_timestamp_column="ValidFrom",
    current_flag_column="CurrentRecord",
    record_creation_timestamp_column="record_available_at",
)
invoice_table = ds.get_source_table(
    database_name="hive_metastore", schema_name="grocery", table_name="groceryinvoice"
).create_event_table(
    name="groceryinvoice",
    event_id_column="GroceryInvoiceGuid",
    event_timestamp_column="Timestamp",
    event_timestamp_timezone_offset_column="tz_offset",
    record_creation_timestamp_column="record_available_at",
)
invoice_table.initialize_default_feature_job_setting()
items_table = ds.get_source_table(
    database_name="hive_metastore", schema_name="grocery", table_name="invoiceitems"
).create_item_table(
    name="invoiceitems",
    event_id_column="GroceryInvoiceGuid",
    item_id_column="GroceryInvoiceItemGuid",
    event_table_name="groceryinvoice",
    record_creation_timestamp_column="record_available_at",
)
product_table = ds.get_source_table(
    database_name="hive_metastore", schema_name="grocery", table_name="groceryproduct_embedding"
).create_dimension_table(name="groceryproduct", dimension_id_column="GroceryProductGuid")

Create and tag entities¶

In [5]:

            
                Copied!
                
                    
                    
                
                

        
# entities
catalog.create_entity(name="grocerycustomer", serving_names=["GROCERYCUSTOMERGUID"])
catalog.create_entity(name="groceryinvoice", serving_names=["GROCERYINVOICEGUID"])
catalog.create_entity(name="groceryitem", serving_names=["GROCERYINVOICEITEMGUID"])
catalog.create_entity(name="groceryproduct", serving_names=["GROCERYPRODUCTGUID"])
catalog.create_entity(name="productgroup", serving_names=["PRODUCTGROUP"])
catalog.create_entity(name="frenchstate", serving_names=["FRENCHSTATE"])

# tag the entities for the grocery customer table
customer_table.GroceryCustomerGuid.as_entity("grocerycustomer")
customer_table.State.as_entity("frenchstate")

# tag the entities for the grocery invoice table
invoice_table.GroceryInvoiceGuid.as_entity("groceryinvoice")
invoice_table.GroceryCustomerGuid.as_entity("grocerycustomer")

# tag the entities for the grocery items table
items_table.GroceryInvoiceItemGuid.as_entity("groceryitem")
items_table.GroceryInvoiceGuid.as_entity("groceryinvoice")
items_table.GroceryProductGuid.as_entity("groceryproduct")

# tag the entities for the grocery product table
product_table.GroceryProductGuid.as_entity("groceryproduct")
product_table.ProductGroup.as_entity("productgroup")
# entities
catalog.create_entity(name="grocerycustomer", serving_names=["GROCERYCUSTOMERGUID"])
catalog.create_entity(name="groceryinvoice", serving_names=["GROCERYINVOICEGUID"])
catalog.create_entity(name="groceryitem", serving_names=["GROCERYINVOICEITEMGUID"])
catalog.create_entity(name="groceryproduct", serving_names=["GROCERYPRODUCTGUID"])
catalog.create_entity(name="productgroup", serving_names=["PRODUCTGROUP"])
catalog.create_entity(name="frenchstate", serving_names=["FRENCHSTATE"])

# tag the entities for the grocery customer table
customer_table.GroceryCustomerGuid.as_entity("grocerycustomer")
customer_table.State.as_entity("frenchstate")

# tag the entities for the grocery invoice table
invoice_table.GroceryInvoiceGuid.as_entity("groceryinvoice")
invoice_table.GroceryCustomerGuid.as_entity("grocerycustomer")

# tag the entities for the grocery items table
items_table.GroceryInvoiceItemGuid.as_entity("groceryitem")
items_table.GroceryInvoiceGuid.as_entity("groceryinvoice")
items_table.GroceryProductGuid.as_entity("groceryproduct")

# tag the entities for the grocery product table
product_table.GroceryProductGuid.as_entity("groceryproduct")
product_table.ProductGroup.as_entity("productgroup")

Get and join views¶

In [6]:

            
                Copied!
                
invoice_view = invoice_table.get_view()
items_view = items_table.get_view()
product_view = product_table.get_view()
invoice_view = invoice_table.get_view()
items_view = items_table.get_view()
product_view = product_table.get_view()

In [7]:

            
                Copied!
                
items_view = items_view.join(product_view)
items_view = items_view.join(invoice_view, rsuffix="_invoice")
items_view = items_view.join(product_view)
items_view = items_view.join(invoice_view, rsuffix="_invoice")

Create invoice product groups embedding¶

In [8]:

            
                Copied!
                
invoice_product_group_embedding = items_view.groupby("GroceryInvoiceGuid").aggregate(
    "ProductGroupEmbedding", 
    method=fb.AggFunc.AVG, 
    feature_name="INVOICE_Avg_of_ProductGroup_Embedding",
)
invoice_product_group_embedding = items_view.groupby("GroceryInvoiceGuid").aggregate(
    "ProductGroupEmbedding", 
    method=fb.AggFunc.AVG, 
    feature_name="INVOICE_Avg_of_ProductGroup_Embedding",
)

Create customer level product group embeddings¶

In [9]:

            
                Copied!
                
                    
                    
                
                

        
customer_avg_product_groups = items_view.groupby("GroceryCustomerGuid").aggregate_over(
    "ProductGroupEmbedding",
    method=fb.AggFunc.AVG,
    feature_names=["CUSTOMER_Avg_of_ProductGroup_Embedding_14d", "CUSTOMER_Avg_of_ProductGroup_Embedding_183d"], 
    windows=["14d", "183d"]
)

customer_avg_product_groups_cosine = customer_avg_product_groups["CUSTOMER_Avg_of_ProductGroup_Embedding_14d"].vec.cosine_similarity(
    customer_avg_product_groups["CUSTOMER_Avg_of_ProductGroup_Embedding_183d"]
)
customer_avg_product_groups_cosine.name = "CUSTOMER_Consistency_of_Avg_of_ProductGroup_Embedding_14d_183d"
customer_avg_product_groups = items_view.groupby("GroceryCustomerGuid").aggregate_over(
    "ProductGroupEmbedding",
    method=fb.AggFunc.AVG,
    feature_names=["CUSTOMER_Avg_of_ProductGroup_Embedding_14d", "CUSTOMER_Avg_of_ProductGroup_Embedding_183d"], 
    windows=["14d", "183d"]
)

customer_avg_product_groups_cosine = customer_avg_product_groups["CUSTOMER_Avg_of_ProductGroup_Embedding_14d"].vec.cosine_similarity(
    customer_avg_product_groups["CUSTOMER_Avg_of_ProductGroup_Embedding_183d"]
)
customer_avg_product_groups_cosine.name = "CUSTOMER_Consistency_of_Avg_of_ProductGroup_Embedding_14d_183d"

Preview features¶

In [10]:

            
                Copied!
                
features = fb.FeatureGroup([
    customer_avg_product_groups_cosine,
    customer_avg_product_groups,
    invoice_product_group_embedding,
])
features = fb.FeatureGroup([
    customer_avg_product_groups_cosine,
    customer_avg_product_groups,
    invoice_product_group_embedding,
])

In [11]:

            
                Copied!
                
                    
                    
                
                

        
observation_table = items_view.create_observation_table(
     name="Preview table",
     sample_rows=10,
     columns=["Timestamp", "GroceryCustomerGuid", "GroceryInvoiceGuid"],
     columns_rename_mapping={
         "Timestamp": "POINT_IN_TIME",
         "GroceryCustomerGuid": "GROCERYCUSTOMERGUID",
         "GroceryInvoiceGuid": "GROCERYINVOICEGUID",
     },
 )
observation_table = items_view.create_observation_table(
     name="Preview table",
     sample_rows=10,
     columns=["Timestamp", "GroceryCustomerGuid", "GroceryInvoiceGuid"],
     columns_rename_mapping={
         "Timestamp": "POINT_IN_TIME",
         "GroceryCustomerGuid": "GROCERYCUSTOMERGUID",
         "GroceryInvoiceGuid": "GROCERYINVOICEGUID",
     },
 )

In [12]:

            
                Copied!
                
features.preview(observation_table.to_pandas())
features.preview(observation_table.to_pandas())

Out[12]:

	POINT_IN_TIME	GROCERYCUSTOMERGUID	GROCERYINVOICEGUID	CUSTOMER_Consistency_of_Avg_of_ProductGroup_Embedding_14d_183d	CUSTOMER_Avg_of_ProductGroup_Embedding_14d	CUSTOMER_Avg_of_ProductGroup_Embedding_183d	INVOICE_Avg_of_ProductGroup_Embedding
0	2023-01-25 16:17:02+00:00	967e4beb-c889-4ff9-9140-66655248bbde	452619d8-da2c-4b87-8869-fabd8617f0b5	0.947687	[-0.07795123751428601, 0.026106864757143, -0.0...	[-0.05989266951781601, 0.028268971296264003, -...	[-0.024409980539999997, 0.0467800437, 0.022734...
1	2023-01-25 17:58:34+00:00	ed957eca-941a-47df-93ee-30ab1ae45ee5	3b7f52a4-eab9-409b-b329-11547eff92a3	NaN	NaN	[-0.050461753724194, 0.03191136753629, -0.0203...	[-0.056538398775, 0.038149968064286, -0.025998...
2	2023-06-22 16:39:56+00:00	7de2b5ba-8078-4c60-aa69-ef7b68c42349	ebe828ea-44af-4865-b2ed-eea02857b889	0.943922	[-0.054724733771875006, 0.04170257735937501, 0...	[-0.054099960254567, 0.036855927644231005, -0....	[-0.0376417868, 0.0208652486275, -0.0210912462...
3	2022-08-01 09:44:05+00:00	52476713-ad96-42e3-983d-31b1d3745a3f	47ec017f-c65b-4bc9-917d-5a7b211c3621	0.866544	[-0.05683987525, 0.045103899862500005, -0.0329...	[-0.053429381090463, 0.018712235034605, -0.015...	[-0.0524657229375, 0.02363832113125, -0.018923...
4	2022-08-16 17:13:05+00:00	e5d71d02-7fec-4eaf-8327-84b3583b7cec	bccd2292-1920-407f-a307-79d6237eab7c	0.956710	[-0.058180038666667, 0.035606893166667, -0.016...	[-0.044520215697211006, 0.031701440613546, -0....	[-0.020503219, 0.03539911665625, -0.0233798176...
5	2022-08-18 15:43:51+00:00	8ad0c1f9-e6cb-43d2-a459-3297f1b29f55	ac5c7e47-8f65-4538-9146-f9edf130ea35	NaN	NaN	[-0.039852984969097005, 0.030179503280556002, ...	[-0.0536430575, 0.037170080630769, -0.01672707...
6	2023-03-14 19:55:25+00:00	00b5a352-7300-4fad-adea-907f8d68f393	79e801ba-0bec-48fa-b505-6da5bc159434	0.756868	[-0.008574453749999999, 0.06710812775, -0.0338...	[-0.050725246419149, 0.036322272146809005, -0....	[-0.054892299269231004, 0.047410778044231, -0....
7	2023-04-05 18:02:15+00:00	1b82b9eb-cc54-4cc4-a7e3-9a7417faa8a5	17745065-0092-4a56-852c-39e8dde94479	0.920511	[-0.05063521202381, 0.033994672869048004, -0.0...	[-0.054906137815789006, 0.037337559488596, -0....	[-0.048078875181818005, 0.015896207727273, -0....
8	2023-04-08 17:43:14+00:00	34be2f38-fe5b-4c18-863d-178b7ad6ff4e	08b15987-d39c-4719-adbb-72e29059c81f	0.955495	[-0.059002682755263006, 0.033196180236842, -0....	[-0.057579961551508005, 0.033872317339907004, ...	[-0.075094953, 0.036157493033333, -0.055885230...
9	2023-04-16 14:53:28+00:00	2b068f1d-d99b-4c2f-a737-46f619a76cc8	512a824c-641a-4e9e-a259-6fbe700e805f	0.861831	[-0.058782766133333005, 0.0196883025, -0.00864...	[-0.051915258536648, 0.02764649977358, -0.0113...	[-0.06907756024999999, 0.01242989875, -0.01187...

In [ ]: