Skip to content

Commit d779567

Browse files
oavdeevoavdeev
authored
Multiple tiny AWS related fixes (#1083)
* multiple tiny AWS related fixes * Fix object path math in client staging uploader * Add random suffix to the historical retrieval output location * EMR job now actually checks for status * add boto3 dep Signed-off-by: Oleg Avdeev <oleg.v.avdeev@gmail.com> * a couple more bugfixes Signed-off-by: Oleg Avdeev <oleg.v.avdeev@gmail.com>
1 parent e426ab8 commit d779567

6 files changed

Lines changed: 21 additions & 11 deletions

File tree

‎infra/docker/jobservice/Dockerfile‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ RUN make compile-protos-python
1717
COPY .git .git
1818
COPY README.md README.md
1919
RUN pip install -e sdk/python -U
20+
RUN pip install "s3fs" "boto3" "urllib3>=1.25.4"
2021

2122
#
2223
# Download grpc_health_probe to run health checks

‎infra/docker/jupyter/Dockerfile‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ RUN pip install -r sdk/python/requirements-ci.txt
1818
COPY .git .git
1919
COPY README.md README.md
2020
RUN pip install -e sdk/python -U
21+
RUN pip install "s3fs" "boto3" "urllib3>=1.25.4"
2122

2223
# Switch back to original user and workdir
2324
USER $NB_UID

‎sdk/python/feast/cli.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -362,7 +362,7 @@ def job():
362362
pass
363363

364364

365-
@job.command(name="start-offline-to-online")
365+
@job.command(name="sync-offline-to-online")
366366
@click.option(
367367
"--feature-table",
368368
"-t",

‎sdk/python/feast/client.py‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -913,7 +913,7 @@ def get_historical_features(
913913
else entity_staging_uri.netloc
914914
)
915915
staging_client.upload_file(
916-
df_export_path.name, bucket, entity_staging_uri.path
916+
df_export_path.name, bucket, entity_staging_uri.path.lstrip("/")
917917
)
918918
entity_source = FileSource(
919919
"event_timestamp",
@@ -923,7 +923,11 @@ def get_historical_features(
923923
)
924924

925925
return start_historical_feature_retrieval_job(
926-
self, entity_source, feature_tables, output_format, output_location
926+
self,
927+
entity_source,
928+
feature_tables,
929+
output_format,
930+
os.path.join(output_location, str(uuid.uuid4())),
927931
)
928932

929933
def get_historical_features_df(

‎sdk/python/feast/pyspark/launchers/aws/emr.py‎

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
JobLauncher,
1515
RetrievalJob,
1616
RetrievalJobParameters,
17+
SparkJobFailure,
1718
SparkJobStatus,
1819
StreamIngestionJob,
1920
StreamIngestionJobParameters,
@@ -83,10 +84,13 @@ def __init__(self, emr_client, job_ref: EmrJobRef, output_file_uri: str):
8384
self._output_file_uri = output_file_uri
8485

8586
def get_output_file_uri(self, timeout_sec=None):
86-
_wait_for_job_state(
87+
state = _wait_for_job_state(
8788
self._emr_client, self._job_ref, TERMINAL_STEP_STATES, timeout_sec
8889
)
89-
return self._output_file_uri
90+
if state in SUCCEEDED_STEP_STATES:
91+
return self._output_file_uri
92+
else:
93+
raise SparkJobFailure("Spark job failed")
9094

9195

9296
class EmrBatchIngestionJob(EmrJobMixin, BatchIngestionJob):
@@ -215,7 +219,7 @@ def historical_feature_retrieval(
215219
return EmrRetrievalJob(
216220
self._emr_client(),
217221
job_ref,
218-
os.path.join(job_params.get_destination_path(), _random_string(8)),
222+
os.path.join(job_params.get_destination_path()),
219223
)
220224

221225
def offline_to_online_ingestion(

‎sdk/python/feast/pyspark/launchers/aws/emr_utils.py‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -258,13 +258,13 @@ def _wait_for_job_state(
258258
job: EmrJobRef,
259259
desired_states: List[str],
260260
timeout_seconds: Optional[int],
261-
):
261+
) -> str:
262262
if job.step_id is None:
263263
step_id = _get_first_step_id(emr_client, job.cluster_id)
264264
else:
265265
step_id = job.step_id
266266

267-
_wait_for_step_state(
267+
return _wait_for_step_state(
268268
emr_client, job.cluster_id, step_id, desired_states, timeout_seconds
269269
)
270270

@@ -290,17 +290,17 @@ def _wait_for_step_state(
290290
step_id: str,
291291
desired_states: List[str],
292292
timeout_seconds: Optional[int],
293-
):
293+
) -> str:
294294
"""
295295
Wait up to timeout seconds for job to go into one of the desired states.
296296
"""
297297
start_time = time.time()
298298
while (timeout_seconds is None) or (time.time() - start_time < timeout_seconds):
299299
state = _get_step_state(emr_client, cluster_id, step_id)
300300
if state in desired_states:
301-
return
301+
return state
302302
else:
303-
time.sleep(0.5)
303+
time.sleep(1)
304304
else:
305305
raise TimeoutError(
306306
f'Timeout waiting for job state to become {"|".join(desired_states)}'

0 commit comments

Comments
 (0)