Feast on Amazon 解决方案

sudo yum install python3-develsudo yum install -y yum-utilssudo yum-config-manager —add-repo https://rpm.releases.hashicorp.com/AmazonLinux/hashicorp.reposudo yum -y install terraform

project: feast_aws_reporegistry: data/registry.dbprovider: awsonline_store:  type: dynamodb  region: ap-southeast-1offline_store:  type: redshift  cluster_id: feast-demo2-redshift-cluster  region: ap-southeast-1  database: flinkstreamdb  user: awsuser  s3_staging_location: s3://feastdemobucket  iam_role: arn:aws:iam::**********:role/s3_spectrum_role

cd infrasudo terraform initsudo terraform plan -var="admin_password=xxxxx"sudo terraform apply -var="admin_password=xxxxx"

aws redshift-data execute-statement \    —region ap-southeast-1 \    —cluster-identifier feast-demo-redshift-cluster \    —db-user awsuser \    —database dev —sql "create external schema spectrum from data catalog database 'flinkstreamdb' iam_role \    'arn:aws:iam::**********:role/s3_spectrum_role' create external database if not exists;“

pip3 install -U numpy==1.21pip3 install feast[aws]

feast init -t xxxxx（repository_name）AWS Region (e.g. us-west-2): ap-southeast-1Redshift Cluster ID: feast-demo-redshift-clusterRedshift Database Name: flinkstreamdbRedshift User Name: awsuserRedshift S3 Staging Location (s3://*): s3://feastdemobucketRedshift IAM Role for S3 (arn:aws:iam::*:role/*): arn:aws:iam::xxxxxx:role/s3_spectrum_role

$ tree ./feast_aws_repo/./feast_aws_repo/├── data│   └── registry.db├── driver_repo.py├── feature_store.yaml

project: feast_aws_reporegistry: data/registry.dbprovider: awsonline_store:  type: dynamodb  region: ap-southeast-1offline_store:  type: redshift  cluster_id: feast-demo2-redshift-cluster  region: ap-southeast-1  database: flinkstreamdb  user: awsuser  s3_staging_location: s3://feastdemobucket  iam_role: arn:aws:iam::xxxxxxx:role/s3_spectrum_role

from datetime import timedeltafrom feast import Entity, Feature, FeatureView, RedshiftSource, ValueTypedriver = Entity(    name="driver_id",    join_key="driver_id",    value_type=ValueType.INT64,)driver_stats_source = RedshiftSource(    table="feast_driver_hourly_stats",    event_timestamp_column="event_timestamp",    created_timestamp_column="created",)
driver_stats_fv = FeatureView(    name="driver_hourly_stats",    entities=["driver_id"],    ttl=timedelta(weeks=52),    features=[        Feature(name="conv_rate", dtype=ValueType.FLOAT),        Feature(name="acc_rate", dtype=ValueType.FLOAT),        Feature(name="avg_daily_trips", dtype=ValueType.INT64),    ],    batch_source=driver_stats_source,    tags={"team": "driver_performance"},)

from datetime import datetime, timedeltaimport pandas as pdfrom feast import FeatureStorefrom driver_repo import driver, driver_stats_fvfs = FeatureStore(repo_path="./")>>> print(fs)<feast.feature_store.FeatureStore object at 0x7f48d47098d0>>>> print(driver_stats_fv){  "spec": {    "name": "driver_hourly_stats",    "entities": [      "driver_id"    ],    "features": [      {        "name": "conv_rate",        "valueType": "FLOAT"      },      {        "name": "acc_rate",        "valueType": "FLOAT"      },      {        "name": "avg_daily_trips",        "valueType": "INT64"      }    ],    "tags": {      "team": "driver_performance"    },    "ttl": "31449600s",    "batchSource": {      "type": "BATCH_REDSHIFT",      "eventTimestampColumn": "event_timestamp",      "createdTimestampColumn": "created",      "redshiftOptions": {        "table": "feast_driver_hourly_stats"      },      "dataSourceClassType": "feast.infra.offline_stores.redshift_source.RedshiftSource"    },    "online": true  },  "meta": {}}

features = ["driver_hourly_stats:conv_rate", "driver_hourly_stats:acc_rate"]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": [1001, 1002, 1003],         }     ) training_df = fs.get_historical_features(         features=features, entity_df=entity_df     ).to_df()

>>> training_df          event_timestamp  driver_id  conv_rate  acc_rate0 2022-07-04 02:33:54.114       1001   0.036082  0.7077441 2022-07-05 14:33:54.114       1002   0.522306  0.9832332 2022-07-07 02:33:54.114       1003   0.734294  0.034062

feast_features = [       "zipcode_features:city",       "zipcode_features:state",       "zipcode_features:location_type",       "zipcode_features:tax_returns_filed",       "zipcode_features:population",       "zipcode_features:total_wages",       "credit_history:credit_card_due",       "credit_history:mortgage_due",       "credit_history:student_loan_due",       "credit_history:vehicle_loan_due",       "credit_history:hard_pulls",       "credit_history:missed_payments_2y",       "credit_history:missed_payments_1y",       "credit_history:missed_payments_6m",       "credit_history:bankruptcies",   ]training_df = self.fs.get_historical_features(           entity_df=entity_df, features=feast_features).to_df()

CURRENT_TIME=$(date -u +"%Y-%m-%dT%H:%M:%S")feast materialize-incremental $CURRENT_TIME
Materializing 1 feature views to 2022-07-07 08:00:03+00:00 into the sqlite onlinestore.driver_hourly_stats from 2022-07-06 16:25:47+00:00 to 2022-07-07 08:00:03+00:00:100%|████████████████████████████████████████████| 5/5 [00:00<00:00, 592.05it/s]

feast materialize 2022-07-13T00:00:00 2022-07-19T00:00:00
Materializing 1 feature views from 2022-07-13 00:00:00+00:00 to 2022-07-19 00:00:00+00:00 into the dynamodb online store.driver_hourly_stats:100%|█████████████████████████████████████████████████████████████████| 5/5 [00:00<00:00, 51.18it/s]

>>> online_features = fs.get_online_features(         features=features, entity_rows=[{"driver_id": 1001}, {"driver_id": 1002}],     ).to_dict()>>> print(pd.DataFrame.from_dict(online_features))   acc_rate  conv_rate  driver_id0  0.179407   0.984951       10011  0.023422   0.069323       1002

project: feast_spark_projectregistry: data/registry.dbprovider: localoffline_store:    type: spark    spark_conf:        spark.master: yarn        spark.ui.enabled: "true"        spark.eventLog.enabled: "true"        spark.sql.catalogImplementation: "hive"        spark.sql.parser.quotedRegexColumnNames: "true"        spark.sql.session.timeZone: "UTC"

source /etc/spark/conf/spark-env.shexport PYTHONPATH="${SPARK_HOME}/python/:$PYTHONPATH"export PYTHONPATH="${SPARK_HOME}/python/lib/py4j-0.10.9-src.zip:$PYTHONPATH"

from feast.infra.offline_stores.contrib.spark_offline_store.spark_source import ( SparkSource,)
driver_hourly_stats= SparkSource(        name="driver_hourly_stats",        query="SELECT event_timestamp as ts, created_timestamp as  created, conv_rate,conv_rate,conv_rate FROM emr_feature_store.driver_hourly_stats",        event_timestamp_column="ts",        created_timestamp_column="created"        )

WITH driver_hourly_stats__entity_dataframe AS (    SELECT        driver_id,        entity_timestamp,        driver_hourly_stats__entity_row_unique_id    FROM entity_dataframe    GROUP BY        driver_id,        entity_timestamp,        driver_hourly_stats__entity_row_unique_id),

        AND subquery.event_timestamp >= entity_dataframe.entity_timestamp - 86400 * interval '1' second        
                AND subquery.driver_id = entity_dataframe.driver_id        ),

            conv_rate as conv_rate,                     acc_rate as acc_rate            FROM (SELECT driver_id,event_timestamp as ts, created_timestamp as  created, conv_rate,acc_rate,avg_daily_trips FROM emr_feature_store.driver_hourly_stats)    WHERE ts <= '2022-07-25T03:27:05.903000'        AND ts >= '2022-07-21T03:27:05.903000'    ),

            INNER JOIN driver_hourly_stats__dedup            USING (driver_hourly_stats__entity_row_unique_id, event_timestamp, created_timestamp)            )    WHERE row_number = 1)