forked from feast-dev/feast
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathstream_processor.py
More file actions
87 lines (70 loc) · 2.83 KB
/
Copy pathstream_processor.py
File metadata and controls
87 lines (70 loc) · 2.83 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
from abc import ABC
from typing import Callable
import pandas as pd
from pyspark.sql import DataFrame
from feast.data_source import DataSource
from feast.importer import import_class
from feast.repo_config import FeastConfigBaseModel
from feast.stream_feature_view import StreamFeatureView
STREAM_PROCESSOR_CLASS_FOR_TYPE = {
("spark", "kafka"): "feast.infra.contrib.spark_kafka_processor.SparkKafkaProcessor",
}
# TODO: support more types other than just Spark.
StreamTable = DataFrame
class ProcessorConfig(FeastConfigBaseModel):
# Processor mode (spark, etc)
mode: str
# Ingestion source (kafka, kinesis, etc)
source: str
class StreamProcessor(ABC):
"""
A StreamProcessor can ingest and transform data for a specific stream feature view,
and persist that data to the online store.
Attributes:
sfv: The stream feature view on which the stream processor operates.
data_source: The stream data source from which data will be ingested.
"""
sfv: StreamFeatureView
data_source: DataSource
def __init__(self, sfv: StreamFeatureView, data_source: DataSource):
self.sfv = sfv
self.data_source = data_source
def ingest_stream_feature_view(self) -> None:
"""
Ingests data from the stream source attached to the stream feature view; transforms the data
and then persists it to the online store.
"""
pass
def _ingest_stream_data(self) -> StreamTable:
"""
Ingests data into a StreamTable.
"""
pass
def _construct_transformation_plan(self, table: StreamTable) -> StreamTable:
"""
Applies transformations on top of StreamTable object. Since stream engines use lazy
evaluation, the StreamTable will not be materialized until it is actually evaluated.
For example: df.collect() in spark or tbl.execute() in Flink.
"""
pass
def _write_to_online_store(self, table: StreamTable) -> None:
"""
Returns query for persisting data to the online store.
"""
pass
def get_stream_processor_object(
config: ProcessorConfig,
sfv: StreamFeatureView,
write_function: Callable[[pd.DataFrame, str, str], None],
):
"""
Returns a stream processor object based on the config mode and stream source type. The write function is a
function that wraps the feature store "write_to_online_store" capability.
"""
if config.mode == "spark" and config.source == "kafka":
stream_processor = STREAM_PROCESSOR_CLASS_FOR_TYPE[("spark", "kafka")]
module_name, class_name = stream_processor.rsplit(".", 1)
cls = import_class(module_name, class_name, "Processor")
return cls(sfv=sfv, config=config, write_function=write_function,)
else:
raise ValueError("other processors besides spark-kafka not supported")