Skip to content

Commit e5903d0

Browse files
committed
stage only local files
Signed-off-by: Oleksii Moskalenko <moskalenko.alexey@gmail.com>
1 parent 934e359 commit e5903d0

1 file changed

Lines changed: 7 additions & 6 deletions

File tree

sdk/python/feast/pyspark/launchers/gcloud/dataproc.py

Lines changed: 7 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -218,20 +218,21 @@ def __init__(
218218
client_options={"api_endpoint": f"{region}-dataproc.googleapis.com:443"}
219219
)
220220

221-
def _stage_files(self, pyspark_script: str, job_id: str) -> str:
221+
def _stage_file(self, file_path: str, job_id: str) -> str:
222+
if not os.path.isfile(file_path):
223+
return file_path
224+
222225
staging_client = get_staging_client("gs")
223-
blob_path = os.path.join(
224-
self.remote_path, job_id, os.path.basename(pyspark_script),
225-
)
226-
staging_client.upload_file(pyspark_script, self.staging_bucket, blob_path)
226+
blob_path = os.path.join(self.remote_path, job_id, os.path.basename(file_path),)
227+
staging_client.upload_file(file_path, self.staging_bucket, blob_path)
227228

228229
return f"gs://{self.staging_bucket}/{blob_path}"
229230

230231
def dataproc_submit(
231232
self, job_params: SparkJobParameters
232233
) -> Tuple[Job, Callable[[], Job], Callable[[], None]]:
233234
local_job_id = str(uuid.uuid4())
234-
main_file_uri = self._stage_files(job_params.get_main_file_path(), local_job_id)
235+
main_file_uri = self._stage_file(job_params.get_main_file_path(), local_job_id)
235236
job_config: Dict[str, Any] = {
236237
"reference": {"job_id": local_job_id},
237238
"placement": {"cluster_name": self.cluster_name},

0 commit comments

Comments
 (0)