Skip to content

Commit f4f345e

Browse files
khorshuhengkhorshuheng
authored
Enhance job api to return associated feature table and start time (#1259)
* Enhance job api to return associated feature table and start time Signed-off-by: Khor Shu Heng <[email protected]> * Fix typo for datetime conversion method Signed-off-by: Khor Shu Heng <[email protected]> * Fix return type and wrong argument Signed-off-by: Khor Shu Heng <[email protected]> * Provide job start time to get historical feature response Signed-off-by: Khor Shu Heng <[email protected]> * Use utc timestamp for start time Signed-off-by: Khor Shu Heng <[email protected]> * Rename fields Signed-off-by: Khor Shu Heng <[email protected]> * Add log uri as fields for retreived job Signed-off-by: Khor Shu Heng <[email protected]> * Cast optional str to str type Signed-off-by: Khor Shu Heng <[email protected]> Co-authored-by: Khor Shu Heng <[email protected]>
1 parent 286941d commit f4f345e

15 files changed

Lines changed: 360 additions & 37 deletions

File tree

‎protos/feast/core/JobService.proto‎

Lines changed: 46 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -73,24 +73,31 @@ message Job {
7373
// Current job status
7474
JobStatus status = 3;
7575
// Deterministic hash of the Job
76-
string hash = 8;
76+
string hash = 4;
77+
// Start time of the Job
78+
google.protobuf.Timestamp start_time = 5;
7779

7880
message RetrievalJobMeta {
79-
string output_location = 4;
81+
string output_location = 1;
8082
}
8183

8284
message OfflineToOnlineMeta {
85+
string table_name = 1;
8386
}
8487

8588
message StreamToOnlineMeta {
89+
string table_name = 1;
8690
}
8791

8892
// JobType specific metadata on the job
8993
oneof meta {
90-
RetrievalJobMeta retrieval = 5;
91-
OfflineToOnlineMeta batch_ingestion = 6;
92-
StreamToOnlineMeta stream_ingestion = 7;
94+
RetrievalJobMeta retrieval = 6;
95+
OfflineToOnlineMeta batch_ingestion = 7;
96+
StreamToOnlineMeta stream_ingestion = 8;
9397
}
98+
99+
// Path to Spark job logs, if available
100+
string log_uri = 9;
94101
}
95102

96103
// Ingest data from offline store into online store
@@ -107,8 +114,17 @@ message StartOfflineToOnlineIngestionJobRequest {
107114
}
108115

109116
message StartOfflineToOnlineIngestionJobResponse {
110-
// Job ID assigned by Feast
111-
string id = 1;
117+
// Job ID assigned by Feast
118+
string id = 1;
119+
120+
// Job start time
121+
google.protobuf.Timestamp job_start_time = 2;
122+
123+
// Feature table associated with the job
124+
string table_name = 3;
125+
126+
// Path to Spark job logs, if available
127+
string log_uri = 4;
112128
}
113129

114130
message GetHistoricalFeaturesRequest {
@@ -136,9 +152,18 @@ message GetHistoricalFeaturesRequest {
136152
}
137153

138154
message GetHistoricalFeaturesResponse {
139-
// Export Job with ID assigned by Feast
140-
string id = 1;
141-
string output_file_uri = 2;
155+
// Export Job with ID assigned by Feast
156+
string id = 1;
157+
158+
// Uri to the join result output file
159+
string output_file_uri = 2;
160+
161+
// Job start time
162+
google.protobuf.Timestamp job_start_time = 3;
163+
164+
// Path to Spark job logs, if available
165+
string log_uri = 4;
166+
142167
}
143168

144169
message StartStreamToOnlineIngestionJobRequest {
@@ -148,8 +173,17 @@ message StartStreamToOnlineIngestionJobRequest {
148173
}
149174

150175
message StartStreamToOnlineIngestionJobResponse {
151-
// Job ID assigned by Feast
152-
string id = 1;
176+
// Job ID assigned by Feast
177+
string id = 1;
178+
179+
// Job start time
180+
google.protobuf.Timestamp job_start_time = 2;
181+
182+
// Feature table associated with the job
183+
string table_name = 3;
184+
185+
// Path to Spark job logs, if available
186+
string log_uri = 4;
153187
}
154188

155189
message ListJobsRequest {

‎sdk/python/feast/client.py‎

Lines changed: 14 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1075,6 +1075,8 @@ def get_historical_features(
10751075
self._extra_grpc_params,
10761076
response.id,
10771077
output_file_uri=response.output_file_uri,
1078+
start_time=response.job_start_time.ToDatetime(),
1079+
log_uri=response.log_uri,
10781080
)
10791081
else:
10801082
return start_historical_feature_retrieval_job(
@@ -1174,7 +1176,12 @@ def start_offline_to_online_ingestion(
11741176
request.end_date.FromDatetime(end)
11751177
response = self._job_service.StartOfflineToOnlineIngestionJob(request)
11761178
return RemoteBatchIngestionJob(
1177-
self._job_service, self._extra_grpc_params, response.id,
1179+
self._job_service,
1180+
self._extra_grpc_params,
1181+
response.id,
1182+
feature_table.name,
1183+
response.job_start_time.ToDatetime(),
1184+
response.log_uri,
11781185
)
11791186

11801187
def start_stream_to_online_ingestion(
@@ -1196,7 +1203,12 @@ def start_stream_to_online_ingestion(
11961203
)
11971204
response = self._job_service.StartStreamToOnlineIngestionJob(request)
11981205
return RemoteStreamIngestionJob(
1199-
self._job_service, self._extra_grpc_params, response.id
1206+
self._job_service,
1207+
self._extra_grpc_params,
1208+
response.id,
1209+
feature_table.name,
1210+
response.job_start_time,
1211+
response.log_uri,
12001212
)
12011213

12021214
def list_jobs(self, include_terminated: bool) -> List[SparkJob]:

‎sdk/python/feast/job_service.py‎

Lines changed: 40 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -5,9 +5,10 @@
55
import time
66
import traceback
77
from concurrent.futures import ThreadPoolExecutor
8-
from typing import Dict, List, Tuple
8+
from typing import Dict, List, Tuple, cast
99

1010
import grpc
11+
from google.protobuf.timestamp_pb2 import Timestamp
1112

1213
import feast
1314
from feast.constants import ConfigOptions as opt
@@ -54,6 +55,7 @@
5455
def _job_to_proto(spark_job: SparkJob) -> JobProto:
5556
job = JobProto()
5657
job.id = spark_job.get_id()
58+
job.log_uri = cast(str, spark_job.get_log_uri() or "")
5759
status = spark_job.get_status()
5860
if status == SparkJobStatus.COMPLETED:
5961
job.status = JobStatus.JOB_STATUS_DONE
@@ -71,11 +73,15 @@ def _job_to_proto(spark_job: SparkJob) -> JobProto:
7173
job.retrieval.output_location = spark_job.get_output_file_uri(block=False)
7274
elif isinstance(spark_job, BatchIngestionJob):
7375
job.type = JobType.BATCH_INGESTION_JOB
76+
job.batch_ingestion.table_name = spark_job.get_feature_table()
7477
elif isinstance(spark_job, StreamIngestionJob):
7578
job.type = JobType.STREAM_INGESTION_JOB
79+
job.stream_ingestion.table_name = spark_job.get_feature_table()
7680
else:
7781
raise ValueError(f"Invalid job type {job}")
7882

83+
job.start_time.FromDatetime(spark_job.get_start_time())
84+
7985
return job
8086

8187

@@ -97,7 +103,16 @@ def StartOfflineToOnlineIngestionJob(
97103
start=request.start_date.ToDatetime(),
98104
end=request.end_date.ToDatetime(),
99105
)
100-
return StartOfflineToOnlineIngestionJobResponse(id=job.get_id())
106+
107+
job_start_timestamp = Timestamp()
108+
job_start_timestamp.FromDatetime(job.get_start_time())
109+
110+
return StartOfflineToOnlineIngestionJobResponse(
111+
id=job.get_id(),
112+
job_start_time=job_start_timestamp,
113+
table_name=request.table_name,
114+
log_uri=job.get_log_uri(),
115+
)
101116

102117
def GetHistoricalFeatures(self, request: GetHistoricalFeaturesRequest, context):
103118
"""Produce a training dataset, return a job id that will provide a file reference"""
@@ -114,8 +129,13 @@ def GetHistoricalFeatures(self, request: GetHistoricalFeaturesRequest, context):
114129

115130
output_file_uri = job.get_output_file_uri(block=False)
116131

132+
job_start_timestamp = Timestamp()
133+
job_start_timestamp.FromDatetime(job.get_start_time())
134+
117135
return GetHistoricalFeaturesResponse(
118-
id=job.get_id(), output_file_uri=output_file_uri
136+
id=job.get_id(),
137+
output_file_uri=output_file_uri,
138+
job_start_time=job_start_timestamp,
119139
)
120140

121141
def StartStreamToOnlineIngestionJob(
@@ -135,7 +155,14 @@ def StartStreamToOnlineIngestionJob(
135155
job_hash = params.get_job_hash()
136156
for job in list_jobs(include_terminated=True, client=self.client):
137157
if isinstance(job, StreamIngestionJob) and job.get_hash() == job_hash:
138-
return StartStreamToOnlineIngestionJobResponse(id=job.get_id())
158+
job_start_timestamp = Timestamp()
159+
job_start_timestamp.FromDatetime(job.get_start_time())
160+
return StartStreamToOnlineIngestionJobResponse(
161+
id=job.get_id(),
162+
job_start_time=job_start_timestamp,
163+
table_name=job.get_feature_table(),
164+
log_uri=job.get_log_uri(),
165+
)
139166
raise RuntimeError(
140167
"Feast Job Service has control loop enabled, but couldn't find the existing stream ingestion job for the given FeatureTable"
141168
)
@@ -147,7 +174,15 @@ def StartStreamToOnlineIngestionJob(
147174
feature_table=feature_table,
148175
extra_jars=[],
149176
)
150-
return StartStreamToOnlineIngestionJobResponse(id=job.get_id())
177+
178+
job_start_timestamp = Timestamp()
179+
job_start_timestamp.FromDatetime(job.get_start_time())
180+
return StartStreamToOnlineIngestionJobResponse(
181+
id=job.get_id(),
182+
job_start_time=job_start_timestamp,
183+
table_name=request.table_name,
184+
log_uri=job.get_log_uri(),
185+
)
151186

152187
def ListJobs(self, request, context):
153188
"""List all types of jobs"""

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

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,18 @@ def cancel(self):
6767
"""
6868
raise NotImplementedError
6969

70+
@abc.abstractmethod
71+
def get_start_time(self) -> datetime:
72+
"""
73+
Get job start time.
74+
"""
75+
76+
def get_log_uri(self) -> Optional[str]:
77+
"""
78+
Get path to Spark job log, if applicable.
79+
"""
80+
return None
81+
7082

7183
class SparkJobParameters(abc.ABC):
7284
@abc.abstractmethod
@@ -496,6 +508,18 @@ class BatchIngestionJob(SparkJob):
496508
Container for the ingestion job result
497509
"""
498510

511+
@abc.abstractmethod
512+
def get_feature_table(self) -> str:
513+
"""
514+
Get the feature table name associated with this job. Return empty string if unable to
515+
determine the feature table, such as when the job is created by the earlier
516+
version of Feast.
517+
518+
Returns:
519+
str: Feature table name
520+
"""
521+
raise NotImplementedError
522+
499523

500524
class StreamIngestionJob(SparkJob):
501525
"""
@@ -513,6 +537,18 @@ def get_hash(self) -> str:
513537
"""
514538
raise NotImplementedError
515539

540+
@abc.abstractmethod
541+
def get_feature_table(self) -> str:
542+
"""
543+
Get the feature table name associated with this job. Return `None` if unable to
544+
determine the feature table, such as when the job is created by the earlier
545+
version of Feast.
546+
547+
Returns:
548+
str: Feature table name
549+
"""
550+
raise NotImplementedError
551+
516552

517553
class JobLauncher(abc.ABC):
518554
"""

0 commit comments

Comments
 (0)