docs: run Feast offline-to-online pipeline with Spark Operator

docs/en/train/guides/assets/feast-offline-to-online-inference/batch.py

@@ -0,0 +1,160 @@
+import copy
+import json
+import os
+import shutil
+import subprocess
+from pathlib import Path
+from urllib.parse import urlparse
+
+import numpy as np
+import pandas as pd
+import yaml
+from feast import FeatureStore
+from pyspark.sql import SparkSession, functions as F
+
+project = os.environ["FEAST_PROJECT"]
+bucket = os.environ["S3_BUCKET"]
+dataset_key = os.environ["S3_DATASET_KEY"].strip("/")
+dataset_uri = f"s3a://{bucket}/{dataset_key}"
+region = os.environ["AWS_DEFAULT_REGION"]
+repo = Path("/tmp/feast-repo")
+model_repo = Path("/mnt/models/repo")
+repo.mkdir(parents=True, exist_ok=True)
+model_repo.mkdir(parents=True, exist_ok=True)
+
+endpoint = urlparse(os.environ["S3_ENDPOINT_URL"])
+endpoint_host = endpoint.netloc or endpoint.path
+ssl_enabled = str(endpoint.scheme == "https").lower()
+spark = (
+    SparkSession.builder.appName("feast-offline-batch")
+    .config("spark.hadoop.fs.s3a.endpoint", endpoint_host)
+    .config("spark.hadoop.fs.s3a.endpoint.region", region)
+    .config("spark.hadoop.fs.s3a.path.style.access", "true")
+    .config("spark.hadoop.fs.s3a.connection.ssl.enabled", ssl_enabled)
+    .getOrCreate()
+)
+
+n_rows = 240
+events = (
+    spark.range(n_rows)
+    .withColumn("driver_id", (F.col("id") % 12 + 1).cast("long"))
+    .withColumn(
+        "event_timestamp",
+        F.timestamp_seconds(F.lit(1767225600) + F.col("id") * 3600),
+    )
+    .withColumn("created", F.col("event_timestamp") + F.expr("INTERVAL 1 MINUTE"))
+    .withColumn("conv_rate", (F.lit(0.25) + F.lit(0.55) * F.rand(7)).cast("float"))
+    .withColumn("acc_rate", (F.lit(0.50) + F.lit(0.45) * F.rand(11)).cast("float"))
+    .withColumn(
+        "avg_daily_trips",
+        F.floor(F.lit(2) + F.lit(18) * F.rand(13)).cast("long"),
+    )
+    .withColumn(
+        "label",
+        (
+            (
+                F.col("conv_rate") * 2
+                + F.col("acc_rate")
+                + F.col("avg_daily_trips") / 20
+            )
+            > 1.8
+        ).cast("long"),
+    )
+    .drop("id")
+)
+events.repartition(4, "driver_id").write.mode("overwrite").partitionBy(
+    "driver_id"
+).parquet(dataset_uri)
+
+client_config = yaml.safe_load(Path("/etc/feast/feature_store.yaml").read_text())
+batch_config = copy.deepcopy(client_config)
+batch_config["offline_store"] = {
+    "type": "spark",
+    "spark_conf": {
+        "spark.sql.session.timeZone": "UTC",
+        "spark.hadoop.fs.s3a.endpoint": endpoint_host,
+        "spark.hadoop.fs.s3a.endpoint.region": region,
+        "spark.hadoop.fs.s3a.path.style.access": "true",
+        "spark.hadoop.fs.s3a.connection.ssl.enabled": ssl_enabled,
+    },
+}
+(repo / "feature_store.yaml").write_text(
+    yaml.safe_dump(batch_config, sort_keys=False)
+)
+(repo / "features.py").write_text(
+    f'''from datetime import timedelta
+from feast import Entity, FeatureService, FeatureView, Field
+from feast.infra.offline_stores.contrib.spark_offline_store.spark_source import SparkSource
+from feast.types import Float32, Int64
+from feast.value_type import ValueType
+
+driver = Entity(name="driver", join_keys=["driver_id"], value_type=ValueType.INT64)
+source = SparkSource(
+    name="driver_stats_source",
+    path="{dataset_uri}",
+    file_format="parquet",
+    timestamp_field="event_timestamp",
+    created_timestamp_column="created",
+)
+view = FeatureView(
+    name="driver_hourly_stats",
+    entities=[driver],
+    ttl=timedelta(days=365),
+    schema=[
+        Field(name="conv_rate", dtype=Float32),
+        Field(name="acc_rate", dtype=Float32),
+        Field(name="avg_daily_trips", dtype=Int64),
+    ],
+    online=True,
+    source=source,
+)
+driver_activity_v1 = FeatureService(name="driver_activity_v1", features=[view])
+'''
+)
+
+subprocess.run(["feast", "--chdir", str(repo), "apply"], check=True)
+store = FeatureStore(repo_path=str(repo))
+entity_df = events.select("driver_id", "event_timestamp", "label").toPandas()
+training_df = store.get_historical_features(
+    entity_df=entity_df,
+    features=[
+        "driver_hourly_stats:conv_rate",
+        "driver_hourly_stats:acc_rate",
+        "driver_hourly_stats:avg_daily_trips",
+    ],
+).to_df().dropna()
+columns = ["conv_rate", "acc_rate", "avg_daily_trips"]
+x = training_df[columns].to_numpy(dtype="float64")
+y = training_df["label"].to_numpy(dtype="float64")
+weights = np.linalg.pinv(np.column_stack([np.ones(len(x)), x])) @ y
+np.savez(
+    "/mnt/models/model.npz",
+    weights=weights,
+    feature_columns=np.array(columns),
+)
+
+end_date = events.agg(F.max("event_timestamp")).first()[0] + pd.Timedelta(hours=1)
+store.materialize_incremental(end_date)
+online = store.get_online_features(
+    features=store.get_feature_service("driver_activity_v1"),
+    entity_rows=[{"driver_id": 1}, {"driver_id": 2}],
+).to_df()
+if len(online) != 2 or online[columns].isna().any().any():
+    raise RuntimeError(f"online feature verification failed: {online}")
+online.to_json("/mnt/models/online-sample.json", orient="records")
+serving_config = copy.deepcopy(client_config)
+serving_config.pop("offline_store", None)
+(model_repo / "feature_store.yaml").write_text(
+    yaml.safe_dump(serving_config, sort_keys=False)
+)
+shutil.copy("/opt/feast-batch/server.py", model_repo / "server.py")
+print(
+    json.dumps(
+        {
+            "historical_rows": len(training_df),
+            "dataset": dataset_uri,
+            "online_rows": len(online),
+        }
+    )
+)
+spark.stop()

docs/en/train/guides/assets/feast-offline-to-online-inference/feature-store.yaml

@@ -0,0 +1,24 @@
+apiVersion: feast.dev/v1
+kind: FeatureStore
+metadata:
+  name: feast-notebook
+  namespace: feast-demo
+spec:
+  feastProject: feast_demo
+  replicas: 1
+  services:
+    onlineStore:
+      persistence:
+        store:
+          type: redis
+          secretRef:
+            name: feast-data-stores
+    registry:
+      local:
+        persistence:
+          store:
+            type: sql
+            secretRef:
+              name: feast-data-stores
+        server: {}
+    ui: {}

docs/en/train/guides/assets/feast-offline-to-online-inference/inference-service.yaml

@@ -0,0 +1,23 @@
+apiVersion: serving.kserve.io/v1beta1
+kind: InferenceService
+metadata:
+  name: feast-online-model
+  namespace: feast-demo
+  annotations:
+    serving.kserve.io/deploymentMode: RawDeployment
+spec:
+  predictor:
+    model:
+      modelFormat:
+        name: feast-numpy
+        version: "1"
+      protocolVersion: v2
+      runtime: feast-numpy-runtime
+      storageUri: pvc://feast-notebook-model
+      resources:
+        requests:
+          cpu: 100m
+          memory: 256Mi
+        limits:
+          cpu: "1"
+          memory: 1Gi

docs/en/train/guides/assets/feast-offline-to-online-inference/model-inspector.yaml

@@ -0,0 +1,18 @@
+apiVersion: v1
+kind: Pod
+metadata:
+  name: feast-model-inspector
+  namespace: feast-demo
+spec:
+  restartPolicy: Never
+  containers:
+    - name: inspector
+      image: FEAST_MODEL_IMAGE_PLACEHOLDER
+      command: [bash, -c, cat /mnt/models/online-sample.json]
+      volumeMounts:
+        - name: model
+          mountPath: /mnt/models
+  volumes:
+    - name: model
+      persistentVolumeClaim:
+        claimName: feast-notebook-model

docs/en/train/guides/assets/feast-offline-to-online-inference/model-pvc.yaml

@@ -0,0 +1,11 @@
+apiVersion: v1
+kind: PersistentVolumeClaim
+metadata:
+  name: feast-notebook-model
+  namespace: feast-demo
+spec:
+  accessModes:
+    - ReadWriteOnce
+  resources:
+    requests:
+      storage: 1Gi

docs/en/train/guides/assets/feast-offline-to-online-inference/namespaces.yaml

@@ -0,0 +1,9 @@
+apiVersion: v1
+kind: Namespace
+metadata:
+  name: feast-demo
+---
+apiVersion: v1
+kind: Namespace
+metadata:
+  name: feast-operator-system

docs/en/train/guides/assets/feast-offline-to-online-inference/server.py

@@ -0,0 +1,66 @@
+import os
+
+import numpy as np
+import uvicorn
+from fastapi import Body, FastAPI
+from feast import FeatureStore
+
+MODEL_NAME = os.getenv("MODEL_NAME", "feast-online-model")
+weights = np.load("/mnt/models/model.npz")["weights"]
+store = FeatureStore(repo_path="/mnt/models/repo")
+feature_service = store.get_feature_service("driver_activity_v1")
+app = FastAPI()
+
+
+@app.get("/v2/health/live")
+@app.get("/v2/health/ready")
+def ready():
+    return {"ready": True}
+
+
+@app.get("/v2/models/{model_name}")
+@app.get("/v2/models/{model_name}/ready")
+def model_ready(model_name: str):
+    return {"name": model_name, "ready": model_name == MODEL_NAME}
+
+
+@app.post("/v2/models/{model_name}/infer")
+def infer(model_name: str, payload: dict = Body(...)):
+    ids = next(
+        item for item in payload["inputs"] if item["name"] == "driver_id"
+    )["data"]
+    rows = [{"driver_id": int(driver_id)} for driver_id in ids]
+    values = store.get_online_features(
+        features=feature_service,
+        entity_rows=rows,
+    ).to_dict()
+
+    def column(name):
+        if name in values:
+            return values[name]
+        return values[next(key for key in values if key.endswith("__" + name))]
+
+    x = np.column_stack(
+        [
+            np.ones(len(ids)),
+            column("conv_rate"),
+            column("acc_rate"),
+            column("avg_daily_trips"),
+        ]
+    )
+    prediction = (x @ weights).astype("float32")
+    return {
+        "model_name": model_name,
+        "outputs": [
+            {
+                "name": "prediction",
+                "shape": [len(ids)],
+                "datatype": "FP32",
+                "data": prediction.tolist(),
+            }
+        ],
+    }
+
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=8080)

docs/en/train/guides/assets/feast-offline-to-online-inference/serving-runtime.yaml

@@ -0,0 +1,35 @@
+apiVersion: serving.kserve.io/v1alpha1
+kind: ServingRuntime
+metadata:
+  name: feast-numpy-runtime
+  namespace: feast-demo
+spec:
+  containers:
+    - name: kserve-container
+      image: FEAST_MODEL_IMAGE_PLACEHOLDER
+      command: [python, /mnt/models/repo/server.py]
+      ports:
+        - containerPort: 8080
+          name: http1
+          protocol: TCP
+      env:
+        - name: MODEL_NAME
+          value: feast-online-model
+      volumeMounts:
+        - name: online-tls
+          mountPath: /tls/online
+          readOnly: true
+        - name: registry-tls
+          mountPath: /tls/registry
+          readOnly: true
+  protocolVersions: [v2]
+  supportedModelFormats:
+    - name: feast-numpy
+      version: "1"
+  volumes:
+    - name: online-tls
+      secret:
+        secretName: feast-feast-notebook-online-tls
+    - name: registry-tls
+      secret:
+        secretName: feast-feast-notebook-registry-tls