@@ -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