tessl/pypi-feast

Python SDK for Feast - an open source feature store for machine learning that manages features for both training and serving environments.

—

Pending

Overview

Eval results

Files

Feature Views

Name: tessl/pypi-feast
Author: tessl

Feature views define how features are computed, stored, and served in Feast. They specify the schema, data sources, entities, and transformation logic for groups of related features. Different feature view types support various feature engineering patterns from batch processing to real-time transformations.

Capabilities

Standard Feature Views

Standard feature views define batch features computed from offline data sources with configurable time-to-live for freshness management.

class FeatureView:
    def __init__(
        self,
        name: str,
        entities: List[Union[Entity, str]],
        schema: List[Field],
        source: DataSource,
        ttl: Optional[timedelta] = None,
        tags: Optional[Dict[str, str]] = None,
        owner: str = "",
        description: str = ""
    ):
        """
        Define a standard feature view for batch features.
        
        Parameters:
        - name: Unique feature view name
        - entities: List of entities this view is associated with
        - schema: List of feature field definitions
        - source: Data source for feature computation
        - ttl: Time-to-live for feature freshness (None for no expiration)
        - tags: Key-value metadata tags
        - owner: Feature view owner/maintainer
        - description: Human-readable description
        """

Batch Feature Views

Batch feature views enable complex feature transformations using SQL or Python for advanced feature engineering workflows.

class BatchFeatureView:
    def __init__(
        self,
        name: str,
        entities: List[Union[Entity, str]],
        schema: List[Field],
        source: DataSource,
        ttl: Optional[timedelta] = None,
        batch_source: Optional[DataSource] = None,
        description: str = "",
        tags: Optional[Dict[str, str]] = None,
        owner: str = ""
    ):
        """
        Define a batch feature view with transformation capabilities.
        
        Parameters:
        - name: Unique feature view name
        - entities: Associated entities for join keys
        - schema: Output feature schema after transformation
        - source: Primary data source
        - ttl: Feature time-to-live
        - batch_source: Optional separate batch processing source
        - description: Feature view description
        - tags: Metadata tags
        - owner: Owner identifier
        """

On-Demand Feature Views

On-demand feature views compute features in real-time during serving using Python transformations applied to request data and existing features.

class OnDemandFeatureView:
    def __init__(
        self,
        name: str,
        sources: Dict[str, Union[FeatureView, FeatureService, RequestSource]],
        schema: List[Field],
        udf: Union[PythonTransformation, str],
        description: str = "",
        tags: Optional[Dict[str, str]] = None,
        owner: str = ""
    ):
        """
        Define an on-demand feature view with real-time transformations.
        
        Parameters:
        - name: Unique feature view name
        - sources: Dictionary mapping source names to feature views or request sources
        - schema: Output schema of transformed features
        - udf: Python transformation function or code string
        - description: Feature view description  
        - tags: Metadata tags
        - owner: Owner identifier
        """

Stream Feature Views

Stream feature views process streaming data sources for real-time feature updates and low-latency serving scenarios.

class StreamFeatureView:
    def __init__(
        self,
        name: str,
        entities: List[Union[Entity, str]],
        schema: List[Field],
        source: Union[KafkaSource, KinesisSource, PushSource],
        ttl: Optional[timedelta] = None,
        tags: Optional[Dict[str, str]] = None,
        owner: str = "",
        description: str = ""
    ):
        """
        Define a stream feature view for real-time data processing.
        
        Parameters:
        - name: Unique feature view name
        - entities: Associated entities
        - schema: Feature schema
        - source: Streaming data source (Kafka, Kinesis, or Push)
        - ttl: Feature time-to-live
        - tags: Metadata tags
        - owner: Owner identifier
        - description: Feature view description
        """

Feature View Operations

Common operations available across all feature view types for inspection and configuration management.

def with_name(self, name: str) -> "FeatureView":
    """Create a copy with a different name."""

def with_projection(self, feature_list: List[str]) -> FeatureViewProjection:
    """Create a projection with selected features."""

@property
def name(self) -> str:
    """Feature view name."""

@property
def entities(self) -> List[str]:
    """Associated entity names."""

@property
def features(self) -> List[Field]:
    """Feature schema fields."""

Field Definitions

Feature schema fields define individual features within feature views including their data types and metadata.

@dataclass
class Field:
    name: str
    dtype: ValueType
    description: str = ""
    tags: Optional[Dict[str, str]] = None

    def __init__(
        self,
        name: str,
        dtype: ValueType,
        description: str = "",
        tags: Optional[Dict[str, str]] = None
    ):
        """
        Define a feature field.
        
        Parameters:
        - name: Feature name
        - dtype: Feature data type (ValueType)
        - description: Feature description
        - tags: Feature-level metadata tags
        """

Usage Examples

Standard Feature View Creation

from feast import FeatureView, Entity, Field, FileSource, ValueType
from datetime import timedelta

# Define entities
driver = Entity(name="driver", value_type=ValueType.INT64)

# Define data source
driver_source = FileSource(
    path="data/driver_stats.parquet",
    timestamp_field="event_timestamp"
)

# Create feature view
driver_hourly_stats = FeatureView(
    name="driver_hourly_stats",
    entities=[driver],
    ttl=timedelta(hours=1),
    schema=[
        Field(
            name="conv_rate", 
            dtype=ValueType.FLOAT,
            description="Driver conversion rate"
        ),
        Field(
            name="acc_rate", 
            dtype=ValueType.FLOAT,
            description="Driver acceptance rate"
        ),
        Field(
            name="avg_daily_trips", 
            dtype=ValueType.INT64,
            description="Average daily trips"
        )
    ],
    source=driver_source,
    description="Hourly driver performance statistics",
    tags={"team": "ml-platform", "domain": "transportation"},
    owner="ml-team@company.com"
)

On-Demand Feature View with Transformations

from feast import OnDemandFeatureView, RequestSource
from feast.transformation.python_transformation import PythonTransformation

# Define request source for real-time data
request_source = RequestSource(
    name="request_data",
    schema=[
        Field(name="val_to_add", dtype=ValueType.INT64),
        Field(name="val_to_add_2", dtype=ValueType.INT64)
    ]
)

# Define transformation function
@feast.on_demand_feature_view(
    sources={"driver_stats": driver_hourly_stats, "input_request": request_source},
    schema=[
        Field(name="conv_rate_plus_val1", dtype=ValueType.DOUBLE),
        Field(name="conv_rate_plus_val2", dtype=ValueType.DOUBLE)
    ]
)
def transformed_conv_rate(inputs: pd.DataFrame) -> pd.DataFrame:
    df = pd.DataFrame()
    df["conv_rate_plus_val1"] = inputs["conv_rate"] + inputs["val_to_add"]
    df["conv_rate_plus_val2"] = inputs["conv_rate"] + inputs["val_to_add_2"] 
    return df

Stream Feature View for Real-time Data

from feast import StreamFeatureView, KafkaSource
from feast.data_format import JsonFormat

# Define Kafka source
kafka_source = KafkaSource(
    kafka_bootstrap_servers="localhost:9092",
    message_format=JsonFormat(schema_json=""),
    topic="driver_events",
    timestamp_field="event_timestamp"
)

# Create stream feature view
driver_stream_features = StreamFeatureView(
    name="driver_activity_stream",
    entities=[driver],
    schema=[
        Field(name="trips_today", dtype=ValueType.INT64),
        Field(name="earnings_today", dtype=ValueType.DOUBLE),
        Field(name="hours_worked", dtype=ValueType.FLOAT)
    ],
    source=kafka_source,
    ttl=timedelta(minutes=30),
    description="Real-time driver activity metrics"
)

Feature View Projections and Selection

# Create feature view projection with selected features
driver_projection = driver_hourly_stats.with_projection([
    "conv_rate",
    "acc_rate"
])

# Use projection in feature service
from feast import FeatureService

driver_ranking_service = FeatureService(
    name="driver_ranking",
    features=[driver_projection]
)

# Create renamed copy
driver_daily_stats = driver_hourly_stats.with_name("driver_daily_stats")

Multi-Entity Feature Views

# Define multiple entities
customer = Entity(name="customer", value_type=ValueType.INT64)
product = Entity(name="product", value_type=ValueType.STRING)

# Multi-entity data source
order_source = FileSource(
    path="data/order_features.parquet",
    timestamp_field="order_timestamp"
)

# Multi-entity feature view
order_features = FeatureView(
    name="customer_product_features",
    entities=[customer, product],
    schema=[
        Field(name="order_count_7d", dtype=ValueType.INT64),
        Field(name="total_spent_7d", dtype=ValueType.DOUBLE),
        Field(name="avg_order_value", dtype=ValueType.DOUBLE)
    ],
    source=order_source,
    ttl=timedelta(days=1),
    description="Customer-product interaction features"
)

Types

class FeatureViewProjection:
    """Projection of selected features from a feature view."""
    
    @property
    def name(self) -> str:
        """Projected feature view name."""
        
    @property  
    def features(self) -> List[Field]:
        """Selected feature fields."""

class PythonTransformation:
    """Python transformation for on-demand feature views."""
    
    def __init__(self, udf: Callable, udf_string: str = ""):
        """
        Parameters:
        - udf: Python function for transformation
        - udf_string: String representation of the function
        """

Install with Tessl CLI