Skip to content

Commit 221badc

Browse files
tsotnettsotnet
committed
Implement half of JobService functionality
Signed-off-by: Tsotne Tabidze <tsotnet@gmail.com>
1 parent 4396d0f commit 221badc

7 files changed

Lines changed: 144 additions & 39 deletions

File tree

‎infra/charts/feast/charts/feast-jobservice/templates/deployment.yaml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,11 @@ spec:
5757
{{- end }}
5858

5959
env:
60+
- name: FEAST_CORE_URL
61+
value: "{{ .Release.Name }}-feast-core:6565"
62+
- name: FEAST_HISTORICAL_SERVING_URL
63+
value: "{{ .Release.Name }}-feast-batch-serving:6566"
64+
6065
{{- if .Values.gcpServiceAccount.enabled }}
6166
- name: GOOGLE_APPLICATION_CREDENTIALS
6267
value: /etc/secrets/google/{{ .Values.gcpServiceAccount.existingSecret.key }}

‎infra/charts/feast/values.yaml‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,10 @@ feast-jupyter:
1313
# feast-jupyter.enabled -- Flag to install Feast Jupyter Notebook with SDK
1414
enabled: true
1515

16+
feast-jobservice:
17+
# feast-jobservice.enabled -- Flag to install Feast Job Service
18+
enabled: true
19+
1620
postgresql:
1721
# postgresql.enabled -- Flag to install Postgresql
1822
enabled: true

‎infra/docker/jobservice/Dockerfile‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
FROM python:3.7-slim-buster
1+
FROM jupyter/pyspark-notebook:ae5f7e104dd5
22

33
USER root
44
WORKDIR /feast
@@ -27,4 +27,10 @@ RUN wget -q https://github.com/grpc-ecosystem/grpc-health-probe/releases/downloa
2727
-O /usr/bin/grpc-health-probe && \
2828
chmod +x /usr/bin/grpc-health-probe
2929

30+
ENV FEAST_SPARK_LAUNCHER standalone
31+
ENV FEAST_SPARK_STANDALONE_MASTER "local[*]"
32+
ENV FEAST_SPARK_HOME $SPARK_HOME
33+
ENV FEAST_SPARK_EXTRA_OPTIONS "--jars https://storage.googleapis.com/hadoop-lib/gcs/gcs-connector-hadoop2-latest.jar \
34+
--conf spark.hadoop.fs.gs.impl=com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem"
35+
3036
CMD ["feast", "server"]

‎protos/feast/core/JobService.proto‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -126,8 +126,8 @@ message StartOfflineToOnlineIngestionJobResponse {
126126
}
127127

128128
message GetHistoricalFeaturesRequest {
129-
// List of features that are being retrieved
130-
repeated feast.serving.FeatureReferenceV2 features = 1;
129+
// List of feature references that are being retrieved
130+
repeated string feature_refs = 1;
131131

132132
// Batch DataSource that can be used to obtain entity values for historical retrieval.
133133
// For each entity value, a feature value will be retrieved for that value/timestamp

‎sdk/python/feast/client.py‎

Lines changed: 66 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
CONFIG_ENABLE_AUTH_KEY,
3434
CONFIG_GRPC_CONNECTION_TIMEOUT_DEFAULT_KEY,
3535
CONFIG_JOB_SERVICE_ENABLE_SSL_KEY,
36+
CONFIG_JOB_SERVICE_ENABLED,
3637
CONFIG_JOB_SERVICE_SERVER_SSL_CERT_KEY,
3738
CONFIG_JOB_SERVICE_URL_KEY,
3839
CONFIG_PROJECT_KEY,
@@ -67,6 +68,11 @@
6768
)
6869
from feast.core.CoreService_pb2_grpc import CoreServiceStub
6970
from feast.core.JobService_pb2_grpc import JobServiceStub
71+
from feast.core.JobService_pb2 import (
72+
GetHistoricalFeaturesRequest,
73+
StartOfflineToOnlineIngestionJobRequest,
74+
StartStreamToOnlineIngestionJobRequest,
75+
)
7076
from feast.data_format import ParquetFormat
7177
from feast.data_source import BigQuerySource, FileSource
7278
from feast.entity import Entity
@@ -190,6 +196,10 @@ def _job_service(self):
190196
191197
Returns: JobServiceStub
192198
"""
199+
# Don't initialize job service stub if the job service is disabled
200+
if self._config.get(CONFIG_JOB_SERVICE_ENABLED) == "False":
201+
return None
202+
193203
if not self._job_service_stub:
194204
channel = create_grpc_channel(
195205
url=self._config.get(CONFIG_JOB_SERVICE_URL_KEY),
@@ -853,8 +863,9 @@ def get_historical_features(
853863
self,
854864
feature_refs: List[str],
855865
entity_source: Union[pd.DataFrame, FileSource, BigQuerySource],
856-
project: str = None,
857-
) -> RetrievalJob:
866+
project: Optional[str] = None,
867+
destination_path: Optional[str] = None,
868+
) -> Union[RetrievalJob, str]:
858869
"""
859870
Launch a historical feature retrieval job.
860871
@@ -873,11 +884,12 @@ def get_historical_features(
873884
retrieval job.
874885
project: Specifies the project that contains the feature tables
875886
which the requested features belong to.
887+
destination_path: Specifies the path in a bucket to write the exported feature data files
876888
877889
Returns:
878-
Returns a retrieval job object that can be used to monitor retrieval
879-
progress asynchronously, and can be used to materialize the
880-
results.
890+
If jobs are launched locally, returns a retrieval job object that can be used to monitor retrieval
891+
progress asynchronously, and can be used to materialize the results.
892+
Otherwise, if jobs are launched through Feast Job Service, returns a job id.
881893
882894
Examples:
883895
>>> from feast import Client
@@ -890,15 +902,6 @@ def get_historical_features(
890902
>>> output_file_uri = feature_retrieval_job.get_output_file_uri()
891903
"gs://some-bucket/output/
892904
"""
893-
feature_tables = self._get_feature_tables_from_feature_refs(
894-
feature_refs, project
895-
)
896-
output_location = os.path.join(
897-
self._config.get(CONFIG_SPARK_HISTORICAL_FEATURE_OUTPUT_LOCATION),
898-
str(uuid.uuid4()),
899-
)
900-
output_format = self._config.get(CONFIG_SPARK_HISTORICAL_FEATURE_OUTPUT_FORMAT)
901-
902905
if isinstance(entity_source, pd.DataFrame):
903906
staging_location = self._config.get(CONFIG_SPARK_STAGING_LOCATION)
904907
entity_staging_uri = urlparse(
@@ -922,13 +925,29 @@ def get_historical_features(
922925
entity_staging_uri.geturl(),
923926
)
924927

925-
return start_historical_feature_retrieval_job(
926-
self,
927-
entity_source,
928-
feature_tables,
929-
output_format,
930-
os.path.join(output_location, str(uuid.uuid4())),
931-
)
928+
if destination_path is None:
929+
destination_path = self._config.get(CONFIG_SPARK_HISTORICAL_FEATURE_OUTPUT_LOCATION)
930+
destination_path = os.path.join(destination_path, str(uuid.uuid4()))
931+
932+
if not self._job_service:
933+
feature_tables = self._get_feature_tables_from_feature_refs(
934+
feature_refs, project
935+
)
936+
output_format = self._config.get(CONFIG_SPARK_HISTORICAL_FEATURE_OUTPUT_FORMAT)
937+
938+
939+
return start_historical_feature_retrieval_job(
940+
self, entity_source, feature_tables, output_format, destination_path
941+
)
942+
else:
943+
request = GetHistoricalFeaturesRequest(
944+
feature_refs=feature_refs,
945+
entities_source=entity_source.to_proto(),
946+
project=project,
947+
destination_path=destination_path,
948+
)
949+
response = self._job_service.GetHistoricalFeatures(request)
950+
return response.id
932951

933952
def get_historical_features_df(
934953
self,
@@ -993,22 +1012,43 @@ def _get_feature_tables_from_feature_refs(
9931012

9941013
def start_offline_to_online_ingestion(
9951014
self, feature_table: FeatureTable, start: datetime, end: datetime,
996-
) -> SparkJob:
1015+
) -> Union[SparkJob, str]:
9971016
"""
9981017
9991018
Launch Ingestion Job from Batch Source to Online Store for given featureTable
10001019
10011020
:param feature_table: FeatureTable which will be ingested
10021021
:param start: lower datetime boundary
10031022
:param end: upper datetime boundary
1004-
:return: Spark Job Proxy object
1023+
:return: Spark Job Proxy object if jobs are launched locally,
1024+
or Spark Job ID if jobs are launched through Feast Job Service
10051025
"""
1006-
return start_offline_to_online_ingestion(feature_table, start, end, self)
1026+
if not self._job_service:
1027+
return start_offline_to_online_ingestion(feature_table, start, end, self)
1028+
else:
1029+
request = StartOfflineToOnlineIngestionJobRequest(
1030+
project=self.project,
1031+
table_name=feature_table.name,
1032+
)
1033+
request.start_date.FromDatetime(start)
1034+
request.end_date.FromDatetime(end)
1035+
response = self._job_service.StartOfflineToOnlineIngestionJob(request)
1036+
return response.id
10071037

10081038
def start_stream_to_online_ingestion(
10091039
self, feature_table: FeatureTable, extra_jars: Optional[List[str]] = None,
1010-
) -> SparkJob:
1011-
return start_stream_to_online_ingestion(feature_table, extra_jars or [], self)
1040+
) -> Union[SparkJob, str]:
1041+
if not self._job_service:
1042+
return start_stream_to_online_ingestion(
1043+
feature_table, extra_jars or [], self
1044+
)
1045+
else:
1046+
request = StartStreamToOnlineIngestionJobRequest(
1047+
project=self.project,
1048+
table_name=feature_table.name,
1049+
)
1050+
response = self._job_service.StartStreamToOnlineIngestionJob(request)
1051+
return response.id
10121052

10131053
def stage_dataframe(
10141054
self,

‎sdk/python/feast/constants.py‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,7 @@ class AuthProvider(Enum):
5252
CONFIG_JOB_SERVICE_URL_KEY = "job_service_url"
5353
CONFIG_JOB_SERVICE_ENABLE_SSL_KEY = "job_service_enable_ssl"
5454
CONFIG_JOB_SERVICE_SERVER_SSL_CERT_KEY = "job_service_server_ssl_cert"
55+
CONFIG_JOB_SERVICE_ENABLED = "job_service_enabled"
5556
CONFIG_GRPC_CONNECTION_TIMEOUT_DEFAULT_KEY = "grpc_connection_timeout_default"
5657
CONFIG_GRPC_CONNECTION_TIMEOUT_APPLY_KEY = "grpc_connection_timeout_apply"
5758
CONFIG_BATCH_FEATURE_REQUEST_WAIT_TIME_SECONDS_KEY = (
@@ -115,7 +116,15 @@ class AuthProvider(Enum):
115116
CONFIG_SERVING_ENABLE_SSL_KEY: "False",
116117
# Path to certificate(s) to secure connection to Feast Serving
117118
CONFIG_SERVING_SERVER_SSL_CERT_KEY: "",
118-
# Default connection timeout to Feast Serving and Feast Core (in seconds)
119+
# Default Feast Job Service URL
120+
CONFIG_JOB_SERVICE_URL_KEY: "localhost:6568",
121+
# Enable or disable TLS/SSL to Feast Job Service
122+
CONFIG_JOB_SERVICE_ENABLE_SSL_KEY: "False",
123+
# Path to certificate(s) to secure connection to Feast Job Service
124+
CONFIG_JOB_SERVICE_SERVER_SSL_CERT_KEY: "",
125+
# Enable or disable Feast Job Service
126+
CONFIG_JOB_SERVICE_ENABLED: "False",
127+
# Default connection timeout to Feast Serving, Feast Core, and Feast Job Service (in seconds)
119128
CONFIG_GRPC_CONNECTION_TIMEOUT_DEFAULT_KEY: "3",
120129
# Default gRPC connection timeout when sending an ApplyFeatureSet command to
121130
# Feast Core (in seconds)

‎sdk/python/feast/job_service.py‎

Lines changed: 50 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,27 @@
33
import grpc
44

55
import feast
6+
from feast.constants import (
7+
CONFIG_SPARK_HISTORICAL_FEATURE_OUTPUT_FORMAT,
8+
CONFIG_SPARK_HISTORICAL_FEATURE_OUTPUT_LOCATION,
9+
)
610
from feast.core import JobService_pb2_grpc
11+
from feast.core.JobService_pb2 import (
12+
GetHistoricalFeaturesResponse,
13+
GetJobResponse,
14+
ListJobsResponse,
15+
StartOfflineToOnlineIngestionJobResponse,
16+
StartStreamToOnlineIngestionJobResponse,
17+
StopJobResponse,
18+
)
19+
from feast.data_source import DataSource
20+
from feast.pyspark.launcher import (
21+
stage_dataframe,
22+
start_historical_feature_retrieval_job,
23+
start_historical_feature_retrieval_spark_session,
24+
start_offline_to_online_ingestion,
25+
start_stream_to_online_ingestion,
26+
)
727
from feast.third_party.grpc.health.v1 import HealthService_pb2_grpc
828
from feast.third_party.grpc.health.v1.HealthService_pb2 import (
929
HealthCheckResponse,
@@ -17,21 +37,42 @@ def __init__(self):
1737

1838
def StartOfflineToOnlineIngestionJob(self, request, context):
1939
"""Start job to ingest data from offline store into online store"""
20-
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
21-
context.set_details("Method not implemented!")
22-
raise NotImplementedError("Method not implemented!")
40+
feature_table = self.client.get_feature_table(
41+
request.table_name, request.project
42+
)
43+
job = start_offline_to_online_ingestion(
44+
feature_table,
45+
request.start_date.ToDatetime(),
46+
request.end_date.ToDatetime(),
47+
self.client,
48+
)
49+
return StartOfflineToOnlineIngestionJobResponse(id=job.get_id())
2350

2451
def GetHistoricalFeatures(self, request, context):
2552
"""Produce a training dataset, return a job id that will provide a file reference"""
26-
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
27-
context.set_details("Method not implemented!")
28-
raise NotImplementedError("Method not implemented!")
53+
feature_tables = self.client._get_feature_tables_from_feature_refs(
54+
request.feature_refs, request.project
55+
)
56+
output_format = self.client._config.get(
57+
CONFIG_SPARK_HISTORICAL_FEATURE_OUTPUT_FORMAT
58+
)
59+
60+
job = start_historical_feature_retrieval_job(
61+
self.client,
62+
DataSource.from_proto(request.entities_source),
63+
feature_tables,
64+
output_format,
65+
request.destination_path,
66+
)
67+
return GetHistoricalFeaturesResponse(id=job.get_id())
2968

3069
def StartStreamToOnlineIngestionJob(self, request, context):
3170
"""Start job to ingest data from stream into online store"""
32-
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
33-
context.set_details("Method not implemented!")
34-
raise NotImplementedError("Method not implemented!")
71+
feature_table = self.client.get_feature_table(
72+
request.table_name, request.project
73+
)
74+
job = start_stream_to_online_ingestion(feature_table, [], self.client)
75+
return StartStreamToOnlineIngestionJobResponse(id=job.get_id())
3576

3677
def ListJobs(self, request, context):
3778
"""List all types of jobs"""

0 commit comments

Comments
 (0)