<jupyter_start><jupyter_text>Unified Dev Environment Generate Raw DataThe data will be used as input for feature generation.<jupyter_code>import pandas as pd
import numpy as np
from datetime import datetime

def generate_entities():
    return np.random.randint(0, 10, size=100).astype(np.int32)

def generate_raw_data(entities):
    df = pd.DataFrame(columns=["driver_id", "ts", "amount", "distance"])
    df['driver_id'] = entities
    df['amount'] = np.random.random(size=100).astype(np.float32) * 500
    df['distance'] = np.random.randint(0, 100, size=100).astype(np.int32)
    df['ts'] = pd.to_datetime(
            np.random.randint(
                datetime(2021, 1, 1).timestamp(),
                datetime(2021, 1, 14).timestamp(),
                size=100),
        unit="s"
    )
    return df

entities = generate_entities()
raw_data = generate_raw_data(entities)
raw_data.to_csv('/home/tzhu/data/test_raw_data.csv', index=False)<jupyter_output><empty_output><jupyter_text>Create Client<jupyter_code>import os
from feast import Client, Feature, Entity, ValueType, FeatureTable
from feast.feature import SQLFeature
from feast.data_source import FileSource, KafkaSource
from feast.data_format import ParquetFormat, AvroFormat

client = Client()
print(client.version())<jupyter_output>{'sdk': {'version': '0.9.0rc2.dev35+g156bd215.d20210217'}, 'serving': 'not configured', 'core': {'url': 'localhost:6565', 'version': '0.10.0-SNAPSHOT'}}<jupyter_text>Inspect Raw Data<jupyter_code>client.sample('/home/tzhu/data/test_raw_data.csv', 10)<jupyter_output>/home/tzhu/.local/lib/python3.8/site-packages/ipykernel/ipkernel.py:283: DeprecationWarning: `should_run_async` will not call `transform_cell` automatically in the future. Please pass the result to `transformed_cell` argument and any exception that happen during thetransform in `preprocessing_exc_tuple` in IPython 7.17 and above.
  and should_run_async(code)<jupyter_text>Define Features and FeatureTables<jupyter_code>feature1 = SQLFeature(name='last_7_day_total_trips', query="select driver_id, count(*) over (partition by driver_id order by ts range between interval '7' day preceding and current row) from driver_stats")
feature2 = SQLFeature(name='last_7_day_avg_amount', query="select driver_id, avg(amount) over (partition by driver_id order by ts range between interval '7' day preceding and current row) from driver_stats")
feature3 = SQLFeature(name='last_7_day_amounts', query="select driver_id, avg(amount) over (partition by driver_id order by ts range between interval '7' day preceding and current row), sum(amount) over (partition by driver_id order by ts range between interval '7' day preceding and current row) from driver_stats")

driver_id = Entity(name="driver_id", description="Driver identifier", value_type=ValueType.INT64)
# client.apply(driver_id)

feature_table = FeatureTable(
    name = "driver_stats",
    entities = ["driver_id"],
    features = [
        feature1,
        feature3,
    ],
    batch_source=FileSource(
        event_timestamp_column="ts",
        file_format=ParquetFormat(),
        file_url='/home/tzhu/data/test_data',
    )
)


# client.apply_feature_table(feature_table)<jupyter_output>/home/tzhu/.local/lib/python3.8/site-packages/ipykernel/ipkernel.py:283: DeprecationWarning: `should_run_async` will not call `transform_cell` automatically in the future. Please pass the result to `transformed_cell` argument and any exception that happen during thetransform in `preprocessing_exc_tuple` in IPython 7.17 and above.
  and should_run_async(code)<jupyter_text>Discover Existing Feature Tables<jupyter_code>print(client.list_feature_tables())
print(client.list_entities())<jupyter_output>[FeatureTable <driver_stats>]
[<feast.entity.Entity object at 0x7ff6023fbc10>]<jupyter_text>Get Offline/Batch Features<jupyter_code>raw_data = client.sample('/home/tzhu/data/test_raw_data.csv', 100)
offline_features = client.get_offline(feature_table, raw_data)
offline_features[:15]<jupyter_output><empty_output><jupyter_text>Get Online/Realtime Features<jupyter_code>client.offline_to_online(feature_table, offline_features)
entities = {'driver_id': [0, 2]}
resp = client.get_online(feature_table, entities)
resp<jupyter_output>select * from driver_stats_features where driver_id in (0, 2) order by ts desc;
/home/tzhu/.local/lib/python3.8/site-packages/ipykernel/ipkernel.py:283: DeprecationWarning: `should_run_async` will not call `transform_cell` automatically in the future. Please pass the result to `transformed_cell` argument and any exception that happen during thetransform in `preprocessing_exc_tuple` in IPython 7.17 and above.
  and should_run_async(code)<jupyter_text>Feature Transformation<jupyter_code>def transform(data: pd.DataFrame) -> pd.DataFrame:
    features = data.copy(deep=True)
    features.pop('ts')
    features.pop('driver_id')
    features['f0'] = offline_features['last_7_day_total_trips'].str[0]
    features['f1'] = offline_features['last_7_day_amounts'].str[0]
    features['f2'] = offline_features['last_7_day_amounts'].str[1]
    features.pop('last_7_day_total_trips')
    features.pop('last_7_day_amounts')
    return features

features = transform(offline_features)<jupyter_output>/home/tzhu/.local/lib/python3.8/site-packages/ipykernel/ipkernel.py:283: DeprecationWarning: `should_run_async` will not call `transform_cell` automatically in the future. Please pass the result to `transformed_cell` argument and any exception that happen during thetransform in `preprocessing_exc_tuple` in IPython 7.17 and above.
  and should_run_async(code)<jupyter_text>Train and Score a Model OfflineBased on the features we have developed, we want to predict if a driver is cheating for a particular trip. With Random Labels<jupyter_code>from sklearn.tree import DecisionTreeClassifier
from sklearn.metrics import roc_curve, roc_auc_score
from matplotlib import pyplot as plt
random_labels = np.random.randint(0, 2, size=100).astype(np.int32)

X_train = features[:70]
y_train = random_labels[:70]
X_test = features[70:]
y_test = random_labels[70:]

clf1 = DecisionTreeClassifier(random_state=0)
clf1.fit(X_train, y_train)
print(clf1.score(X_test, y_test))

predictions = clf1.predict(X_test)
print(roc_auc_score(y_test, predictions))
fpr, tpr, _ = roc_curve(y_test, predictions)

plt.clf()
plt.plot(fpr, tpr)
plt.xlabel('FPR')
plt.ylabel('TPR')
plt.title('ROC curve')
plt.show()<jupyter_output>0.4666666666666667
0.4666666666666666<jupyter_text>With Meaningful Labels<jupyter_code>labels = []
for item in features.itertuples():
    amount = item[1]
    distance = item[2]
    avg_amount = item[4]
    if amount > avg_amount and distance > 20:
        labels.append(1)
    else:
        labels.append(0)

X_train = features[:70]
y_train = labels[:70]
X_test = features[70:]
y_test = labels[70:]

clf2 = DecisionTreeClassifier(random_state=0)
clf2.fit(X_train, y_train)
clf2.score(X_test, y_test)

predictions = clf2.predict(X_test)
print(roc_auc_score(y_test, predictions))
fpr, tpr, _ = roc_curve(y_test, predictions)

plt.clf()
plt.plot(fpr, tpr)
plt.xlabel('FPR')
plt.ylabel('TPR')
plt.title('ROC curve')
plt.show()<jupyter_output>/home/tzhu/.local/lib/python3.8/site-packages/ipykernel/ipkernel.py:283: DeprecationWarning: `should_run_async` will not call `transform_cell` automatically in the future. Please pass the result to `transformed_cell` argument and any exception that happen during thetransform in `preprocessing_exc_tuple` in IPython 7.17 and above.
  and should_run_async(code)
0.9210526315789473<jupyter_text>Online Prediction<jupyter_code>entities = {'driver_id': [0, 2]}
resp = client.get_online(feature_table, entities)
print(clf1.predict(transform(resp)))
print(clf2.predict(transform(resp)))<jupyter_output>select * from driver_stats_features where driver_id in (0, 2) order by ts desc;
[0 1]
[0 0]
/home/tzhu/.local/lib/python3.8/site-packages/ipykernel/ipkernel.py:283: DeprecationWarning: `should_run_async` will not call `transform_cell` automatically in the future. Please pass the result to `transformed_cell` argument and any exception that happen during thetransform in `preprocessing_exc_tuple` in IPython 7.17 and above.
  and should_run_async(code)
{'fraction_non_alphanumeric': 0.1109082055028001, 'fraction_numerical': 0.031896761626491354, 'mean_word_length': 6.131076388888889, 'pattern_counts': {'":': 0, '<': 38, '<?xml version=': 0, '>': 41, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '11', 'dir': 'jupyter-structured-clean-dedup', 'id': '613857', 'license': 'Apache-2.0', 'n_tokens_mistral': 2901, 'n_tokens_neox': 2722, 'n_words': 626, 'path': 'examples/minimal/demo.ipynb', 'repo_name': 'tianshizz/feast', 'size': '8214'}