Paulescu

# Base image from which you build your container
FROM python:3.11.3-slim-buster

# Environment setup
ENV PYTHONUNBUFFERED=1
WORKDIR /app

# Copy application code and install dependencies
COPY . /app
RUN python3 -m pip install -r requirements.txt

Paulescu

my-ml-project
├── requirements.txt
└── src
    ├── api.py
    └── train.py

Paulescu

app = Application.Quix(
  consumer_group="trades_consumer_group",
  auto_offset_reset="earliest",
  auto_create_topics=True,
)

Paulescu

from datetime import date, timedelta
import streamlit as st

# https://docs.streamlit.io/library/advanced-features/caching#minimal-example
# If the function code and parameters (in this case `day`) do not change
# the function output is cached and re-used.
@st.cache_data
def fetch_data_from_external_api(day: date) -> pd.DataFrame:
    """
    Fetches data for the given `day` and returns pandas dataframe

Paulescu

# instantiate dataflow object
from bytewax.dataflow import Dataflow
flow = Dataflow()

# read input data from websocket
flow.input("input", ManualInputConfig(input_builder))

# parse string data to Python dictionaries
# 1-to-1 stateless operation
flow.map(json.loads)

Paulescu

# connect to your feature view
feature_view = feature_store.get_feature_view(
    name='user_engagement_metrics_fv',
    version=1
)

from datetime import date, timedelta
yesterday = date.today() - timedelta(days=1)

# download batch of features from last 24 hours

Paulescu

# connect to your feature view
feature_view = feature_store.get_feature_view(
    name='user_engagement_metrics_fv',
    version=1
)

# download training data from the feature store
training_data, _ = feature_view.training_data()

Paulescu

feature_store.create_feature_view(
    name='user_engagement_metrics_fv',
    version=1,
    query=feature_group.select_all()
)

Paulescu

# fetch data from data warehouse, for example
user_engagement_metrics : pd.DataFrame = get_user_metrics()

# push feature to the feature store
feature_group.insert(user_engagement_metrics)

Paulescu

feature_group = feature_store.create_feature_group(
    name='user_engagement_metrics_fg',
    version=1,
    description="Daily user-level engagement metrics",
    primary_key = ['user_id'],
)