aligned.worker.StreamWorker

class documentation

class StreamWorker: (source)

Constructors: StreamWorker.from_object(repo, file, obj), StreamWorker.from_reference(source, sink_source, views_to_process)

View In Hierarchy

Undocumented

Static Method from_object Undocumented
Static Method from_reference Creates a stream worker.
Method expose_metrics_at Undocumented
Method feature_views_by_topic Undocumented
Method prune_unused_features Undocumented
Method read_from Undocumented
Method read_from_timestamps Undocumented
Async Method start Undocumented
Class Variable repo_definition Undocumented
Class Variable sink_source Undocumented
Class Variable views_to_process Undocumented
Instance Variable default_start_timestamp Undocumented
Instance Variable metric_logging_port Undocumented
Instance Variable read_timestamps Undocumented
Instance Variable should_prune_unused_features Undocumented
@staticmethod
def from_object(repo: Path, file: Path, obj: str) -> StreamWorker: (source)

Undocumented

@staticmethod
def from_reference(source: AsRepoDefinition, sink_source: WritableFeatureSource, views_to_process: set[str] | None = None) -> StreamWorker: (source)

Creates a stream worker.

This object can start a background worker process of streaming data.

Args:

source (StorageFileReference): The storage of the feature store file sink_source (WritableFeatureSource): Where to store the processed features views_to_process (set[str] | None, optional): The views to process.

Defaults to None aka, all streaming views.
Returns:
StreamWorker | None: A worker that can start processing
def expose_metrics_at(self, port: int) -> StreamWorker: (source)

Undocumented

def feature_views_by_topic(self, store: ContractStore) -> dict[str, list[FeatureViewStore]]: (source)

Undocumented

def prune_unused_features(self, should_prune_unused_features: bool | None = None) -> StreamWorker: (source)

Undocumented

def read_from(self, timestamp: str) -> StreamWorker: (source)

Undocumented

def read_from_timestamps(self, timestamps: dict[str, str]) -> StreamWorker: (source)

Undocumented

async def start(self): (source)

Undocumented

repo_definition: AsRepoDefinition = (source)

Undocumented

sink_source: WritableFeatureSource = (source)

Undocumented

views_to_process: set[str] | None = (source)

Undocumented

default_start_timestamp: str | None = (source)

Undocumented

metric_logging_port: int | None = (source)

Undocumented

read_timestamps: dict[str, str] = (source)

Undocumented

should_prune_unused_features: bool = (source)

Undocumented