Source code for feast.batch_feature_view

import warnings
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple, Union

from feast import flags_helper
from feast.data_source import DataSource
from feast.entity import Entity
from feast.feature_view import FeatureView
from feast.field import Field
from feast.protos.feast.core.DataSource_pb2 import DataSource as DataSourceProto

warnings.simplefilter("once", RuntimeWarning)

SUPPORTED_BATCH_SOURCES = {
    "BigQuerySource",
    "FileSource",
    "RedshiftSource",
    "SnowflakeSource",
    "SparkSource",
    "TrinoSource",
    "AthenaSource",
}


[docs]class BatchFeatureView(FeatureView): """ A batch feature view defines a logical group of features that has only a batch data source. Attributes: name: The unique name of the batch feature view. entities: List of entities or entity join keys. ttl: The amount of time this group of features lives. A ttl of 0 indicates that this group of features lives forever. Note that large ttl's or a ttl of 0 can result in extremely computationally intensive queries. schema: The schema of the feature view, including feature, timestamp, and entity columns. If not specified, can be inferred from the underlying data source. source: The batch source of data where this group of features is stored. online: A boolean indicating whether online retrieval is enabled for this feature view. description: A human-readable description. tags: A dictionary of key-value pairs to store arbitrary metadata. owner: The owner of the batch feature view, typically the email of the primary maintainer. """ name: str entities: List[str] ttl: Optional[timedelta] source: DataSource schema: List[Field] entity_columns: List[Field] features: List[Field] online: bool description: str tags: Dict[str, str] owner: str timestamp_field: str materialization_intervals: List[Tuple[datetime, datetime]] def __init__( self, *, name: str, source: DataSource, entities: Optional[Union[List[Entity], List[str]]] = None, ttl: Optional[timedelta] = None, tags: Optional[Dict[str, str]] = None, online: bool = True, description: str = "", owner: str = "", schema: Optional[List[Field]] = None, ): if not flags_helper.is_test(): warnings.warn( "Batch feature views are experimental features in alpha development. " "Some functionality may still be unstable so functionality can change in the future.", RuntimeWarning, ) if ( type(source).__name__ not in SUPPORTED_BATCH_SOURCES and source.to_proto().type != DataSourceProto.SourceType.CUSTOM_SOURCE ): raise ValueError( f"Batch feature views need a batch source, expected one of {SUPPORTED_BATCH_SOURCES} " f"or CUSTOM_SOURCE, got {type(source).__name__}: {source.name} instead " ) super().__init__( name=name, entities=entities, ttl=ttl, tags=tags, online=online, description=description, owner=owner, schema=schema, source=source, )