generate_project_name,

def init_command(project_directory, minimal: bool, template: str):

"""Create a new Feast repository"""

if not project_directory:

project_directory = generate_project_name()

if template and minimal:

from colorama import Fore, Style

click.echo(

f"Please select either a {Style.BRIGHT + Fore.GREEN}template{Style.RESET_ALL} or "

f"{Style.BRIGHT + Fore.GREEN}minimal{Style.RESET_ALL}, not both"

)

exit(1)

if minimal:

template = "minimal"

init_repo(project_directory, template)

file_dir = self._filepath.parent

file_dir.mkdir(exist_ok=True)

from datetime import timedelta

from importlib.abc import Loader

sys.dont_write_bytecode = True

sys.dont_write_bytecode = False

def init_repo(repo_name: str, template: str):

import os

from distutils.dir_util import copy_tree

from pathlib import Path

from colorama import Fore, Style

repo_path = Path(os.path.join(Path.cwd(), repo_name))

repo_path.mkdir(exist_ok=True)

repo_config_path = repo_path / "feature_store.yaml"

if repo_config_path.exists():

new_directory = os.path.relpath(repo_path, os.getcwd())

f"The directory {Style.BRIGHT + Fore.GREEN}{new_directory}{Style.RESET_ALL} contains an existing feature "

f"store repository that may cause a conflict"

print()

sys.exit(1)

# Copy template directory

template_path = str(Path(Path(__file__).parent / "templates" / template).absolute())

if not os.path.exists(template_path):

raise IOError(f"Could not find template {template}")

copy_tree(template_path, str(repo_path))

# Seed the repository

bootstrap_path = repo_path / "bootstrap.py"

if os.path.exists(bootstrap_path):

import importlib.util

spec = importlib.util.spec_from_file_location("bootstrap", str(bootstrap_path))

bootstrap = importlib.util.module_from_spec(spec)

assert isinstance(spec.loader, Loader)

spec.loader.exec_module(bootstrap)

bootstrap.bootstrap() # type: ignore

os.remove(bootstrap_path)

# Template the feature_store.yaml file

feature_store_yaml_path = repo_path / "feature_store.yaml"

replace_str_in_file(

feature_store_yaml_path, "project: my_project", f"project: {repo_name}"

)

# Remove the __pycache__ folder if it exists

import shutil

shutil.rmtree(repo_path / "__pycache__", ignore_errors=True)

import click

click.echo()

click.echo(

f"Creating a new Feast repository in {Style.BRIGHT + Fore.GREEN}{repo_path}{Style.RESET_ALL}."

)

click.echo()

def replace_str_in_file(file_path, match_str, sub_str):

with open(file_path, "r") as f:

contents = f.read()

contents = contents.replace(match_str, sub_str)

with open(file_path, "wt") as f:

f.write(contents)

Each folder in this module is a template that comes packaged with Feast.

* A template is installed with `feast init -t template_name`

* The template name provided during `init` maps directly to the folder name

* It is possible to provide a bootstrap.py script with a template. The script must provide a bootstrap() function. This

script will automatically be executed and can be used to set up data or sources for the user.

* The feature_store.yaml will have its `project` name templated based on the project name provided by the user. The

default project name should be `my_project`. If a different name is chosen then no templating will occur.

from datetime import timedelta

from feast import BigQuerySource, Entity, Feature, FeatureView, ValueType

driver = Entity(

# Name of the entity. Must be unique within a project

name="driver",

# The join key of an entity describes the storage level field/column on which

# features can be looked up. The join key is also used to join feature

# tables/views when building feature vectors

join_key="driver_id",

# The storage level type for an entity

value_type=ValueType.INT64,

)

driver_stats_source = BigQuerySource(

# The BigQuery table where features can be found

table_ref="feast-oss.demo_data.driver_stats",

# The event timestamp is used for point-in-time joins and for ensuring only

# features within the TTL are returned

event_timestamp_column="datetime",

# The (optional) created timestamp is used to ensure there are no duplicate

# feature rows in the offline store or when building training datasets

created_timestamp_column="created",

)

driver_stats_fv = FeatureView(

# The unique name of this feature view. Two feature views in a single

# project cannot have the same name

name="driver_stats",

# The list of entities specifies the keys required for joining or looking

# up features from this feature view. The reference provided in this field

# correspond to the name of a defined entity (or entities)

entities=["driver"],

# The timedelta is the maximum age that each feature value may have

# relative to its lookup time. For historical features (used in training),

# TTL is relative to each timestamp provided in the entity dataframe.

# TTL also allows for eviction of keys from online stores and limits the

# amount of historical scanning required for historical feature values

# during retrieval

ttl=timedelta(weeks=52),

# The list of features defined below act as a schema to both define features

# for both materialization of features into a store, and are used as references

# during retrieval for building a training dataset or serving features

features=[

Feature(name="conv_rate", dtype=ValueType.FLOAT),

Feature(name="acc_rate", dtype=ValueType.FLOAT),

Feature(name="avg_daily_trips", dtype=ValueType.INT64),

],

# Inputs are used to find feature values. In the case of this feature

# view we will query a source table on BigQuery for driver statistics

# features

input=driver_stats_source,

# Tags are user defined key/value pairs that are attached to each

# feature view

tags={"team": "driver_performance"},

)

project: my_project

registry: data/registry.db

provider: gcp

from datetime import datetime, timedelta

import pandas as pd

from driver_repo import driver, driver_stats_fv

from feast import FeatureStore

def main():

pd.set_option("display.max_columns", None)

pd.set_option("display.width", 1000)

# Load the feature store from the current path

fs = FeatureStore(repo_path=".")

# Deploy the feature store to GCP

print("Deploying feature store to GCP...")

fs.apply([driver, driver_stats_fv])

# Select features

feature_refs = ["driver_stats:conv_rate", "driver_stats:acc_rate"]

# Create an entity dataframe. This is the dataframe that will be enriched with historical features

entity_df = pd.DataFrame(

{

"event_timestamp": [

pd.Timestamp(dt, unit="ms", tz="UTC").round("ms")

for dt in pd.date_range(

start=datetime.now() - timedelta(days=3),

end=datetime.now(),

periods=3,

)

],

"driver_id": [5001, 5002, 5003],

}

)

print("Retrieving training data...")

# Retrieve historical features by joining the entity dataframe to the BigQuery table source

training_df = fs.get_historical_features(

feature_refs=feature_refs, entity_df=entity_df

).to_df()

print()

print(training_df)

print()

print("Loading features into the online store...")

fs.materialize_incremental(end_date=datetime.now())

print()

print("Retrieving online features...")

# Retrieve features from the online store (Firestore)

online_features = fs.get_online_features(

feature_refs=feature_refs, entity_rows=[{"driver": 5001}, {"driver": 5002}]

).to_dict()

print()

print(pd.DataFrame.from_dict(online_features))

if __name__ == "__main__()":

main()

def bootstrap():

# Bootstrap() will automatically be called from the init_repo() during `feast init`

import pathlib

from datetime import datetime, timedelta

from feast.driver_test_data import create_driver_hourly_stats_df

repo_path = pathlib.Path(__file__).parent.absolute()

data_path = repo_path / "data"

data_path.mkdir(exist_ok=True)

end_date = datetime.now().replace(microsecond=0, second=0, minute=0)

start_date = end_date - timedelta(days=15)

driver_entities = [1001, 1002, 1003, 1004, 1005]

driver_df = create_driver_hourly_stats_df(driver_entities, start_date, end_date)

driver_stats_path = data_path / "driver_stats.parquet"

driver_df.to_parquet(path=str(driver_stats_path), allow_truncated_timestamps=True)

example_py_file = repo_path / "example.py"

replace_str_in_file(example_py_file, "%PARQUET_PATH%", str(driver_stats_path))

def replace_str_in_file(file_path, match_str, sub_str):

with open(file_path, "r") as f:

contents = f.read()

contents = contents.replace(match_str, sub_str)

with open(file_path, "wt") as f:

f.write(contents)

if __name__ == "__main__":

bootstrap()