Skip to content

Commit 61a6079

Browse files
mike0svmike0sv
committed
emr job and fix import error by rename
Signed-off-by: mike0sv <mike0sv@gmail.com>
1 parent 0485841 commit 61a6079

6 files changed

Lines changed: 25 additions & 3 deletions

File tree

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,7 @@ def _get_optional(option):
6161
new_cluster_template_path=_get_optional(opt.EMR_CLUSTER_TEMPLATE_PATH),
6262
staging_location=config.get(opt.SPARK_STAGING_LOCATION),
6363
emr_log_location=config.get(opt.EMR_LOG_LOCATION),
64+
additional_options=_parse_additional_spark_options(config),
6465
)
6566

6667

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

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -145,6 +145,7 @@ def __init__(
145145
new_cluster_template_path: Optional[str],
146146
staging_location: str,
147147
emr_log_location: str,
148+
additional_options: Dict[str, str],
148149
):
149150
"""
150151
Initialize a dataproc job controller client, used internally for job submission and result
@@ -162,6 +163,8 @@ def __init__(
162163
An S3 staging location for artifacts.
163164
emr_log_location:
164165
S3 location for EMR logs.
166+
additional_options:
167+
Additional configuration options for Spark job
165168
"""
166169

167170
assert existing_cluster_id or new_cluster_template_path
@@ -177,6 +180,7 @@ def __init__(
177180
self._staging_location = staging_location
178181
self._emr_log_location = emr_log_location
179182
self._region = region
183+
self._additional_options = additional_options
180184

181185
def _emr_client(self):
182186

@@ -230,6 +234,7 @@ def historical_feature_retrieval(
230234

231235
step = _historical_retrieval_step(
232236
pyspark_script_path,
237+
conf=self._additional_options,
233238
args=job_params.get_arguments(),
234239
output_file_uri=job_params.get_destination_path(),
235240
)
@@ -260,6 +265,7 @@ def offline_to_online_ingestion(
260265
step = _sync_offline_to_online_step(
261266
jar_s3_path,
262267
ingestion_job_params.get_feature_table_name(),
268+
self._additional_options,
263269
args=ingestion_job_params.get_arguments(),
264270
)
265271

@@ -293,6 +299,7 @@ def start_stream_to_online_ingestion(
293299
jar_s3_path,
294300
extra_jar_paths,
295301
ingestion_job_params.get_feature_table_name(),
302+
self._additional_options,
296303
args=ingestion_job_params.get_arguments(),
297304
job_hash=job_hash,
298305
)

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

Lines changed: 16 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -135,7 +135,7 @@ def _upload_jar(jar_s3_prefix: str, local_path: str) -> str:
135135

136136

137137
def _sync_offline_to_online_step(
138-
jar_path: str, feature_table_name: str, args: List[str],
138+
jar_path: str, feature_table_name: str, conf: Dict[str, str], args: List[str],
139139
) -> Dict[str, Any]:
140140

141141
return {
@@ -155,6 +155,7 @@ def _sync_offline_to_online_step(
155155
"spark-submit",
156156
"--class",
157157
"feast.ingestion.IngestionJob",
158+
*_prepare_conf_args(conf),
158159
"--packages",
159160
"com.google.cloud.spark:spark-bigquery-with-dependencies_2.12:0.17.2",
160161
jar_path,
@@ -344,8 +345,17 @@ def _upload_dataframe(s3prefix: str, df: pandas.DataFrame) -> str:
344345
)
345346

346347

348+
def _prepare_conf_args(conf: Dict[str, str]):
349+
return [
350+
_ for name, value in conf.items() for _ in ["--conf", f'"{name}"="{value}"']
351+
]
352+
353+
347354
def _historical_retrieval_step(
348-
pyspark_script_path: str, args: List[str], output_file_uri: str,
355+
pyspark_script_path: str,
356+
conf: Dict[str, str],
357+
args: List[str],
358+
output_file_uri: str,
349359
) -> Dict[str, Any]:
350360

351361
return {
@@ -361,7 +371,8 @@ def _historical_retrieval_step(
361371
"Value": output_file_uri,
362372
},
363373
],
364-
"Args": ["spark-submit", pyspark_script_path] + args,
374+
"Args": ["spark-submit", *_prepare_conf_args(conf), pyspark_script_path]
375+
+ args,
365376
"Jar": "command-runner.jar",
366377
},
367378
}
@@ -371,6 +382,7 @@ def _stream_ingestion_step(
371382
jar_path: str,
372383
extra_jar_paths: List[str],
373384
feature_table_name: str,
385+
conf: Dict[str, str],
374386
args: List[str],
375387
job_hash: str,
376388
) -> Dict[str, Any]:
@@ -395,6 +407,7 @@ def _stream_ingestion_step(
395407
{"Key": "feast.step_metadata.job_hash", "Value": job_hash},
396408
],
397409
"Args": ["spark-submit", "--class", "feast.ingestion.IngestionJob"]
410+
+ _prepare_conf_args(conf)
398411
+ jars_args
399412
+ [
400413
"--packages",

sdk/python/tests/pyspark/__init__.py renamed to sdk/python/tests/test_pyspark/__init__.py

File renamed without changes.

sdk/python/tests/pyspark/test_launchers.py renamed to sdk/python/tests/test_pyspark/test_launchers.py

File renamed without changes.

‎tests/integration/fixtures/launchers.py‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,4 +20,5 @@ def dataproc_launcher(pytestconfig) -> DataprocClusterLauncher:
2020
executor_instances=executor_instances,
2121
executor_cores=executor_cores,
2222
executor_memory=executor_memory,
23+
additional_options={}
2324
)

0 commit comments

Comments
 (0)