Skip to content

Commit 065b310

Browse files
pyalexkhorshuhengoavdeev
authored
"Start Offline-to-online ingestion" method in Python SDK (#1051)
* launcher move common job properties to job classes Signed-off-by: Oleksii Moskalenko <moskalenko.alexey@gmail.com> universal test * split SparkJob into 2 classes Signed-off-by: Oleksii Moskalenko <moskalenko.alexey@gmail.com> * better name Signed-off-by: Oleksii Moskalenko <moskalenko.alexey@gmail.com> * weak link to client Signed-off-by: Oleksii Moskalenko <moskalenko.alexey@gmail.com> * weak link to client Signed-off-by: Oleksii Moskalenko <moskalenko.alexey@gmail.com> * skip e2e test Signed-off-by: Oleksii Moskalenko <moskalenko.alexey@gmail.com> * Update job naming Co-authored-by: Oleg Avdeev <oleg.v.avdeev@gmail.com> Signed-off-by: Khor Shu Heng <khor.heng@gojek.com> * Remove unused job_id argument Signed-off-by: Khor Shu Heng <khor.heng@gojek.com> * Remove unused import Signed-off-by: Khor Shu Heng <khor.heng@gojek.com> Co-authored-by: Khor Shu Heng <32997938+khorshuheng@users.noreply.github.com> Co-authored-by: Oleg Avdeev <oleg.v.avdeev@gmail.com> Co-authored-by: Khor Shu Heng <khor.heng@gojek.com>
1 parent fe202c9 commit 065b310

9 files changed

Lines changed: 500 additions & 98 deletions

File tree

‎sdk/python/feast/client.py‎

Lines changed: 12 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@
1414
import logging
1515
import multiprocessing
1616
import shutil
17-
import uuid
17+
from datetime import datetime
1818
from itertools import groupby
1919
from typing import Any, Dict, List, Optional, Union
2020

@@ -74,10 +74,11 @@
7474
_write_partitioned_table_from_source,
7575
)
7676
from feast.online_response import OnlineResponse, _infer_online_entity_rows
77-
from feast.pyspark.abc import RetrievalJob
77+
from feast.pyspark.abc import RetrievalJob, SparkJob
7878
from feast.pyspark.launcher import (
7979
start_historical_feature_retrieval_job,
8080
start_historical_feature_retrieval_spark_session,
81+
start_offline_to_online_ingestion,
8182
)
8283
from feast.serving.ServingService_pb2 import (
8384
GetFeastServingInfoRequest,
@@ -817,10 +818,9 @@ def get_historical_features(
817818
CONFIG_SPARK_HISTORICAL_FEATURE_OUTPUT_LOCATION
818819
)
819820
output_format = self._config.get(CONFIG_SPARK_HISTORICAL_FEATURE_OUTPUT_FORMAT)
820-
job_id = f"historical-feature-{str(uuid.uuid4())}"
821821

822822
return start_historical_feature_retrieval_job(
823-
self, entity_source, feature_tables, output_format, output_location, job_id
823+
self, entity_source, feature_tables, output_format, output_location
824824
)
825825

826826
def get_historical_features_df(
@@ -883,3 +883,11 @@ def _get_feature_tables_from_feature_refs(
883883
]
884884
feature_tables.append(feature_table)
885885
return feature_tables
886+
887+
def start_offline_to_online_ingestion(
888+
self,
889+
feature_table: Union[FeatureTable, str],
890+
start: Union[datetime, str],
891+
end: Union[datetime, str],
892+
) -> SparkJob:
893+
return start_offline_to_online_ingestion(feature_table, start, end, self) # type: ignore

‎sdk/python/feast/constants.py‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,7 +67,10 @@ class AuthProvider(Enum):
6767
# Spark Job Config
6868
CONFIG_SPARK_LAUNCHER = "spark_launcher" # standalone, dataproc, emr
6969

70+
CONFIG_SPARK_INGESTION_JOB_JAR = "spark_ingestion_jar"
71+
7072
CONFIG_SPARK_STANDALONE_MASTER = "spark_standalone_master"
73+
CONFIG_SPARK_HOME = "spark_home"
7174

7275
CONFIG_SPARK_DATAPROC_CLUSTER_NAME = "dataproc_cluster_name"
7376
CONFIG_SPARK_DATAPROC_PROJECT = "dataproc_project"
@@ -109,4 +112,6 @@ class AuthProvider(Enum):
109112
CONFIG_MAX_WAIT_INTERVAL_KEY: "60",
110113
# Authentication Provider - Google OpenID/OAuth
111114
CONFIG_AUTH_PROVIDER: "google",
115+
CONFIG_SPARK_LAUNCHER: "dataproc",
116+
CONFIG_SPARK_INGESTION_JOB_JAR: "gs://feast-jobs/feast-ingestion-spark-0.8-SNAPSHOT.jar",
112117
}

‎sdk/python/feast/pyspark/abc.py‎

Lines changed: 150 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,9 @@
11
import abc
2-
from typing import Dict, List
2+
import json
3+
import os
4+
from datetime import datetime
5+
from enum import Enum
6+
from typing import Dict, List, Optional
37

48

59
class SparkJobFailure(Exception):
@@ -10,6 +14,12 @@ class SparkJobFailure(Exception):
1014
pass
1115

1216

17+
class SparkJobStatus(Enum):
18+
IN_PROGRESS = 1
19+
FAILED = 2
20+
COMPLETED = 3
21+
22+
1323
class SparkJob(abc.ABC):
1424
"""
1525
Base class for all spark jobs
@@ -25,6 +35,85 @@ def get_id(self) -> str:
2535
"""
2636
raise NotImplementedError
2737

38+
@abc.abstractmethod
39+
def get_status(self) -> SparkJobStatus:
40+
"""
41+
Job Status retrieval
42+
43+
:return: SparkJobStatus
44+
"""
45+
raise NotImplementedError
46+
47+
48+
class SparkJobParameters(abc.ABC):
49+
@abc.abstractmethod
50+
def get_name(self) -> str:
51+
"""
52+
Getter for job name
53+
:return: Job name
54+
"""
55+
raise NotImplementedError
56+
57+
@abc.abstractmethod
58+
def get_main_file_path(self) -> str:
59+
"""
60+
Getter for jar | python path
61+
:return: Full path to file
62+
"""
63+
raise NotImplementedError
64+
65+
def get_class_name(self) -> Optional[str]:
66+
"""
67+
Getter for main class name if it's applicable
68+
:return: java class path, e.g. feast.ingestion.IngestionJob
69+
"""
70+
return None
71+
72+
@abc.abstractmethod
73+
def get_arguments(self) -> List[str]:
74+
"""
75+
Getter for job arguments
76+
E.g., ["--source", '{"kafka":...}', ...]
77+
:return: List of arguments
78+
"""
79+
raise NotImplementedError
80+
81+
82+
class RetrievalJobParameters(SparkJobParameters):
83+
def __init__(
84+
self,
85+
feature_tables: List[Dict],
86+
feature_tables_sources: List[Dict],
87+
entity_source: Dict,
88+
destination: Dict,
89+
**kwargs,
90+
):
91+
self._feature_tables = feature_tables
92+
self._feature_tables_sources = feature_tables_sources
93+
self._entity_source = entity_source
94+
self._destination = destination
95+
96+
def get_name(self) -> str:
97+
all_feature_tables_names = [ft["name"] for ft in self._feature_tables]
98+
return f"HistoryRetrieval-{'-'.join(all_feature_tables_names)}"
99+
100+
def get_main_file_path(self) -> str:
101+
return os.path.join(
102+
os.path.dirname(__file__), "historical_feature_retrieval_job.py"
103+
)
104+
105+
def get_arguments(self) -> List[str]:
106+
return [
107+
"--feature-tables",
108+
json.dumps(self._feature_tables),
109+
"--feature-tables-sources",
110+
json.dumps(self._feature_tables_sources),
111+
"--entity-source",
112+
json.dumps(self._entity_source),
113+
"--destination",
114+
json.dumps(self._destination),
115+
]
116+
28117

29118
class RetrievalJob(SparkJob):
30119
"""
@@ -53,8 +142,53 @@ def get_output_file_uri(self, timeout_sec=None):
53142
raise NotImplementedError
54143

55144

145+
class IngestionJobParameters(SparkJobParameters):
146+
def __init__(
147+
self,
148+
feature_table: Dict,
149+
source: Dict,
150+
start: datetime,
151+
end: datetime,
152+
jar: str,
153+
**kwargs,
154+
):
155+
self._feature_table = feature_table
156+
self._source = source
157+
self._start = start
158+
self._end = end
159+
self._jar = jar
160+
161+
def get_name(self) -> str:
162+
return (
163+
f"BatchIngestion-{self._feature_table['name']}-"
164+
f"{self._start.strftime('%Y-%m-%d')}-{self._end.strftime('%Y-%m-%d')}"
165+
)
166+
167+
def get_main_file_path(self) -> str:
168+
return self._jar
169+
170+
def get_class_name(self) -> Optional[str]:
171+
return "feast.ingestion.IngestionJob"
172+
173+
def get_arguments(self) -> List[str]:
174+
return [
175+
"--mode",
176+
"offline",
177+
"--feature-table",
178+
json.dumps(self._feature_table),
179+
"--source",
180+
json.dumps(self._source),
181+
"--start",
182+
self._start.strftime("%Y-%m-%dT%H:%M:%S"),
183+
"--end",
184+
self._end.strftime("%Y-%m-%dT%H:%M:%S"),
185+
]
186+
187+
56188
class IngestionJob(SparkJob):
57-
pass
189+
"""
190+
Container for the ingestion job result
191+
"""
58192

59193

60194
class JobLauncher(abc.ABC):
@@ -65,26 +199,21 @@ class JobLauncher(abc.ABC):
65199
@abc.abstractmethod
66200
def historical_feature_retrieval(
67201
self,
68-
pyspark_script: str,
69202
entity_source_conf: Dict,
70203
feature_tables_sources_conf: List[Dict],
71204
feature_tables_conf: List[Dict],
72205
destination_conf: Dict,
73-
job_id: str,
74206
**kwargs,
75207
) -> RetrievalJob:
76208
"""
77209
Submits a historical feature retrieval job to a Spark cluster.
78210
79211
Args:
80-
pyspark_script (str): Local file path to the pyspark script for historical feature
81-
retrieval.
82212
entity_source_conf (Dict): Entity data source configuration.
83213
feature_tables_sources_conf (List[Dict]): List of feature tables data sources configurations.
84214
feature_tables_conf (List[Dict]): List of feature table specification.
85215
The order of the feature table must correspond to that of feature_tables_sources.
86216
destination_conf (Dict): Retrieval job output destination.
87-
job_id (str): A job id that is unique for each job submission.
88217
89218
Raises:
90219
SparkJobFailure: The spark job submission failed, encountered error
@@ -191,3 +320,17 @@ def historical_feature_retrieval(
191320
str: file uri to the result file.
192321
"""
193322
raise NotImplementedError
323+
324+
@abc.abstractmethod
325+
def offline_to_online_ingestion(
326+
self,
327+
jar_path: str,
328+
source_conf: Dict,
329+
feature_table_conf: Dict,
330+
start: datetime,
331+
end: datetime,
332+
) -> IngestionJob:
333+
"""
334+
Submits a batch ingestion job to a Spark cluster.
335+
"""
336+
raise NotImplementedError

0 commit comments

Comments
 (0)