From 604ef8779f78addceeb810ed6800851d0418ec35 Mon Sep 17 00:00:00 2001 From: Chen Xu Date: Thu, 9 Jun 2022 04:20:57 +0800 Subject: [PATCH 1/4] #211 Enable using maven package --- feathr_project/feathr/constants.py | 1 + .../feathr/spark_provider/.gitignore | 1 + .../spark_provider/_databricks_submission.py | 6 +++- .../spark_provider/_synapse_submission.py | 34 +++++++++++++----- .../feathr/spark_provider/noop-1.0.jar | Bin 0 -> 1736 bytes 5 files changed, 32 insertions(+), 10 deletions(-) create mode 100644 feathr_project/feathr/spark_provider/.gitignore create mode 100644 feathr_project/feathr/spark_provider/noop-1.0.jar diff --git a/feathr_project/feathr/constants.py b/feathr_project/feathr/constants.py index 13adb785b..20d57675a 100644 --- a/feathr_project/feathr/constants.py +++ b/feathr_project/feathr/constants.py @@ -24,3 +24,4 @@ TYPEDEF_ARRAY_DERIVED_FEATURE=f"array" TYPEDEF_ARRAY_ANCHOR_FEATURE=f"array" +FEATHR_JAR_MAVEN_REPO="com.linkedin.feathr:feathr_2.12:0.4.0" \ No newline at end of file diff --git a/feathr_project/feathr/spark_provider/.gitignore b/feathr_project/feathr/spark_provider/.gitignore new file mode 100644 index 000000000..ba64b52e6 --- /dev/null +++ b/feathr_project/feathr/spark_provider/.gitignore @@ -0,0 +1 @@ +!noop-1.0.jar \ No newline at end of file diff --git a/feathr_project/feathr/spark_provider/_databricks_submission.py b/feathr_project/feathr/spark_provider/_databricks_submission.py index 3eca8a3a1..579e00527 100644 --- a/feathr_project/feathr/spark_provider/_databricks_submission.py +++ b/feathr_project/feathr/spark_provider/_databricks_submission.py @@ -143,7 +143,11 @@ def submit_feathr_job(self, job_name: str, main_jar_path: str, main_class_name: submission_params['new_cluster']['spark_conf'] = configuration submission_params['new_cluster']['custom_tags'] = job_tags # the feathr main jar file is anyway needed regardless it's pyspark or scala spark - submission_params['libraries'][0]['jar'] = self.upload_or_get_cloud_path(main_jar_path) + if main_jar_path is None: + logger.info("Main JAR file is not set, using default package from Maven") + submission_params['libraries'][0]['maven'] = { "coordinates": FEATHR_JAR_MAVEN_REPO } + else: + submission_params['libraries'][0]['jar'] = self.upload_or_get_cloud_path(main_jar_path) # see here for the submission parameter definition https://docs.microsoft.com/en-us/azure/databricks/dev-tools/api/2.0/jobs#--request-structure-6 if python_files: # this is a pyspark job. definition here: https://docs.microsoft.com/en-us/azure/databricks/dev-tools/api/2.0/jobs#--sparkpythontask diff --git a/feathr_project/feathr/spark_provider/_synapse_submission.py b/feathr_project/feathr/spark_provider/_synapse_submission.py index adfa8e973..e5079d5e7 100644 --- a/feathr_project/feathr/spark_provider/_synapse_submission.py +++ b/feathr_project/feathr/spark_provider/_synapse_submission.py @@ -1,4 +1,6 @@ +from copy import deepcopy import os +import pathlib import re import time import urllib.request @@ -43,7 +45,8 @@ class _FeathrSynapseJobLauncher(SparkJobLauncher): """ Submits spark jobs to a Synapse spark cluster. """ - def __init__(self, synapse_dev_url: str, pool_name: str, datalake_dir: str, executor_size: str, executors: int, credential = None): + + def __init__(self, synapse_dev_url: str, pool_name: str, datalake_dir: str, executor_size: str, executors: int, credential=None): # use DeviceCodeCredential if EnvironmentCredential is not available self.credential = credential # use the same credential for authentication to avoid further login. @@ -60,9 +63,11 @@ def upload_or_get_cloud_path(self, local_path_or_http_path: str): Supports transferring file from an http path to cloud working storage, or upload directly from a local storage. """ logger.info('Uploading {} to cloud..', local_path_or_http_path) - res_path = self._datalake.upload_file_to_workdir(local_path_or_http_path) + res_path = self._datalake.upload_file_to_workdir( + local_path_or_http_path) - logger.info('{} is uploaded to location: {}', local_path_or_http_path, res_path) + logger.info('{} is uploaded to location: {}', + local_path_or_http_path, res_path) return res_path def download_result(self, result_path: str, local_folder: str): @@ -73,7 +78,7 @@ def download_result(self, result_path: str, local_folder: str): return self._datalake.download_file(result_path, local_folder) def submit_feathr_job(self, job_name: str, main_jar_path: str = None, main_class_name: str = None, arguments: List[str] = None, - python_files: List[str]= None, reference_files_path: List[str] = None, job_tags: Dict[str, str] = None, + python_files: List[str] = None, reference_files_path: List[str] = None, job_tags: Dict[str, str] = None, configuration: Dict[str, str] = None): """ Submits the feathr job @@ -96,7 +101,18 @@ def submit_feathr_job(self, job_name: str, main_jar_path: str = None, main_clas job_tags (str): tags of the job, for exmaple you might want to put your user ID, or a tag with a certain information configuration (Dict[str, str]): Additional configs for the spark job """ - assert main_jar_path, 'main_jar_path should not be none or empty but it is none or empty.' + + # Use an no-op jar as the main executable, as we must set the `main_file` in order to submit a Spark job to Azure Synapse + cfg = configuration.copy() # We don't want to mess up input parameters + if main_jar_path is None: + logger.info("Main JAR file is not set, using default package from Maven") + if "spark.jars.packages" in cfg: + cfg["spark.jars.packages"] = ",".join( + [cfg["spark.jars.packages"], FEATHR_JAR_MAVEN_REPO]) + else: + cfg["spark.jars.packages"] = FEATHR_JAR_MAVEN_REPO + current_dir = pathlib.Path(__file__).parent.resolve() + main_jar_path = os.path.join(current_dir, "noop-1.0.jar") if main_jar_path.startswith('abfs'): main_jar_cloud_path = main_jar_path logger.info( @@ -120,7 +136,7 @@ def submit_feathr_job(self, job_name: str, main_jar_path: str = None, main_clas arguments=arguments, reference_files=reference_files_path, tags=job_tags, - configuration=configuration) + configuration=cfg) logger.info('See submitted job here: https://web.azuresynapse.net/en-us/monitoring/sparkapplication') return self.current_job_info @@ -319,7 +335,7 @@ def __init__(self, datalake_dir, credential=None): self.dir_client = self.file_system_client.get_directory_client('/') self.datalake_dir = datalake_dir + \ - '/' if datalake_dir[-1] != '/' else datalake_dir + '/' if datalake_dir[-1] != '/' else datalake_dir def upload_file_to_workdir(self, src_file_path: str) -> str: """ @@ -394,7 +410,7 @@ def download_file(self, target_adls_directory: str, local_dir_cache: str): for folder in result_folders: folder_name = basename(folder) file_in_folder = [os.path.join(folder_name, basename(file_path.name)) for file_path in self.file_system_client.get_paths( - path=folder, recursive=False) if not file_path.is_directory] + path=folder, recursive=False) if not file_path.is_directory] local_paths = [os.path.join(local_dir_cache, file_name) for file_name in file_in_folder] self._download_file_list(local_paths, file_in_folder, directory_client) @@ -405,7 +421,7 @@ def download_file(self, target_adls_directory: str, local_dir_cache: str): self._download_file_list(local_paths, result_paths, directory_client) logger.info('Finish downloading files from {} to {}.', - target_adls_directory,local_dir_cache) + target_adls_directory, local_dir_cache) def _download_file_list(self, local_paths: List[str], result_paths, directory_client): ''' diff --git a/feathr_project/feathr/spark_provider/noop-1.0.jar b/feathr_project/feathr/spark_provider/noop-1.0.jar new file mode 100644 index 0000000000000000000000000000000000000000..24f7cb77584e78fb2be1ad89c6464a28fb324ce2 GIT binary patch literal 1736 zcmWIWW@h1H009<-lObRRl;8x?zOEsTx}JV+`TRsgZ*3~+9=KSU$gDb`lo)+nNojal9t?R_W{$xqm6fx}s zDiu5Dco`RamCV&zD>BJA=(Cnr<<2?XLBifa!3UE1SUVjOS??d&bj5}>*_PYZxuZ`c zcKwu`394dMY#)yubK4WMCx!vxya4%=A?8vGmc0QwE18jj0pvHo{QLsF;YFA4 z29sx+8=tJJ*|(Q@$FayRhUJEAhHWh`+%m7;5o)^lbdtcmneLu1o_d|WnHTJ_Tfp_~ zIR*2jE2baPTE746hRUGxdnTUmx#<^~-*Tzt^F^)rZt;m%ox*SK+2|D>yk9Tp=+B4t z=}uPW*3qY`pH36rZc(_B&uPofS3FMtm;VS@lX~+3lkUI%YTXH^joj84r6uLXzja*y zPsZ5oo{EfN86TsH$4p_F>f=+7+cob$`XGE_LDRbD^ZdW=hldGf%<>~gS#DxkY92g( zF%?T=D$XxT*GsKP%q_@C#iLOHzs5XZq`-A^Ffag<)szm&Z(P7Q(`RB}P$xonL4K}Y zMQ%=OV089j1&P}8>pAwhPtN66jreUQu_b2%ceUte-O?l6S(E?#iOlup(*1n4_~*Ij z6Z6tcrzl(K_r@(|yS{=&*GIi{zuSEI^H;fYBOKQn?^ke=yn1W>MLU+;LW_QINEM$x zDz~A~wzB%}MD>{Cny!_0Coi3gi_5ue^6;ghq};1riuZTFt)8=>LUk*D40pt>y*af> z+uc4)xc0|<-=5_&YNsF6dReeXdD@{BMzTEdJ2|ICc&*!@^tvjY@7vPkLYcOi+SUxJ z=?30AxIeXb3oz#PEKOldIl8j&noRJq*9|wewq`J#*E7BFM<{X1RA!4eXAg2FXnW-e zm-jaAmOI4zxbuYegr|J2yX5xI|I3KT2B6HWf39M2GB6mEfLM>n;4CQ0FGwva$xJN{ zF7rR@t?Q-Z=^N;=D&&K&j`!KeL3}4QLO*Do4K~;iVqmb*$Y6t!L3jZlU!Z}Jfzcx^ zXht^$=5sIoQ~J6ed^Vod_qy)A*5i!s3D2iqI$9@vynXZ-xt>3LE?&S0OreZSBFwl; zAE29oK!D+`BZx*Si(uNYmqsAPAh4tnB@rQ&O$duQ7_b!>5d9z%jscm7;sc=%GoOGm z6$A({{QrLon2g~LhGj9BvoSS;G8sZMJ23sDX-3U*xD8T(nF9>A|NkBE8wAaQFw?MS zLYRR|8taHK4Vonp0mT8XSJ0Cp!qBtKc>M}?BFIog!VK_cWdrHw075gMlWJH%JOJD_ B7~TK? literal 0 HcmV?d00001 From dc6066c420e598cdeb1a30ea0c5dea4a2e369e0c Mon Sep 17 00:00:00 2001 From: Chen Xu Date: Thu, 9 Jun 2022 14:04:26 +0800 Subject: [PATCH 2/4] #211 Enable using maven package --- feathr_project/feathr/constants.py | 2 +- .../spark_provider/_databricks_submission.py | 4 +- .../spark_provider/_synapse_submission.py | 64 +++++++++++------- .../feathr/spark_provider/noop-1.0.jar | Bin 1736 -> 2605 bytes 4 files changed, 44 insertions(+), 26 deletions(-) diff --git a/feathr_project/feathr/constants.py b/feathr_project/feathr/constants.py index 20d57675a..8884753f5 100644 --- a/feathr_project/feathr/constants.py +++ b/feathr_project/feathr/constants.py @@ -24,4 +24,4 @@ TYPEDEF_ARRAY_DERIVED_FEATURE=f"array" TYPEDEF_ARRAY_ANCHOR_FEATURE=f"array" -FEATHR_JAR_MAVEN_REPO="com.linkedin.feathr:feathr_2.12:0.4.0" \ No newline at end of file +FEATHR_MAVEN_ARTIFACT="com.linkedin.feathr:feathr_2.12:0.4.0" \ No newline at end of file diff --git a/feathr_project/feathr/spark_provider/_databricks_submission.py b/feathr_project/feathr/spark_provider/_databricks_submission.py index 579e00527..114e0728a 100644 --- a/feathr_project/feathr/spark_provider/_databricks_submission.py +++ b/feathr_project/feathr/spark_provider/_databricks_submission.py @@ -143,9 +143,9 @@ def submit_feathr_job(self, job_name: str, main_jar_path: str, main_class_name: submission_params['new_cluster']['spark_conf'] = configuration submission_params['new_cluster']['custom_tags'] = job_tags # the feathr main jar file is anyway needed regardless it's pyspark or scala spark - if main_jar_path is None: + if main_jar_path is None or main_jar_path=="": logger.info("Main JAR file is not set, using default package from Maven") - submission_params['libraries'][0]['maven'] = { "coordinates": FEATHR_JAR_MAVEN_REPO } + submission_params['libraries'][0]['maven'] = { "coordinates": FEATHR_MAVEN_ARTIFACT } else: submission_params['libraries'][0]['jar'] = self.upload_or_get_cloud_path(main_jar_path) # see here for the submission parameter definition https://docs.microsoft.com/en-us/azure/databricks/dev-tools/api/2.0/jobs#--request-structure-6 diff --git a/feathr_project/feathr/spark_provider/_synapse_submission.py b/feathr_project/feathr/spark_provider/_synapse_submission.py index e5079d5e7..77d836e8f 100644 --- a/feathr_project/feathr/spark_provider/_synapse_submission.py +++ b/feathr_project/feathr/spark_provider/_synapse_submission.py @@ -97,32 +97,45 @@ def submit_feathr_job(self, job_name: str, main_jar_path: str = None, main_clas job_name (str): name of the job main_jar_path (str): main file paths, usually your main jar file main_class_name (str): name of your main class - arguments (str): all the arugments you want to pass into the spark job - job_tags (str): tags of the job, for exmaple you might want to put your user ID, or a tag with a certain information + arguments (str): all the arguments you want to pass into the spark job + job_tags (str): tags of the job, for example you might want to put your user ID, or a tag with a certain information configuration (Dict[str, str]): Additional configs for the spark job """ - # Use an no-op jar as the main executable, as we must set the `main_file` in order to submit a Spark job to Azure Synapse cfg = configuration.copy() # We don't want to mess up input parameters - if main_jar_path is None: - logger.info("Main JAR file is not set, using default package from Maven") - if "spark.jars.packages" in cfg: - cfg["spark.jars.packages"] = ",".join( - [cfg["spark.jars.packages"], FEATHR_JAR_MAVEN_REPO]) + if not main_jar_path: + # We don't have the main jar, use Maven + if not python_files: + # This is a JAR job + # Azure Synapse/Livy doesn't allow JAR job starts from Maven directly, we must have a jar file uploaded. + # so we have to use a dummy jar as the main file. + logger.info("Main JAR file is not set, using default package from Maven") + # Add Maven dependency to the job + if "spark.jars.packages" in cfg: + cfg["spark.jars.packages"] = ",".join( + [cfg["spark.jars.packages"], FEATHR_MAVEN_ARTIFACT]) + else: + cfg["spark.jars.packages"] = FEATHR_MAVEN_ARTIFACT + # Use the no-op jar as the main file + # This is a dummy jar which contains only one `org.example.Noop` class with one empty `main` function which does nothing + current_dir = pathlib.Path(__file__).parent.resolve() + main_jar_path = os.path.join(current_dir, "noop-1.0.jar") else: - cfg["spark.jars.packages"] = FEATHR_JAR_MAVEN_REPO - current_dir = pathlib.Path(__file__).parent.resolve() - main_jar_path = os.path.join(current_dir, "noop-1.0.jar") - if main_jar_path.startswith('abfs'): - main_jar_cloud_path = main_jar_path - logger.info( - 'Cloud path {} is used for running the job: {}', main_jar_path, job_name) - else: - logger.info('Uploading jar from {} to cloud for running job: {}', - main_jar_path, job_name) - main_jar_cloud_path = self._datalake.upload_file_to_workdir(main_jar_path) - logger.info('{} is uploaded to {} for running job: {}', - main_jar_path, main_jar_cloud_path, job_name) + # This is a PySpark job + pass + main_jar_cloud_path = None + if main_jar_path: + # Now we have a main jar, either feathr or noop + if main_jar_path.startswith('abfs'): + main_jar_cloud_path = main_jar_path + logger.info( + 'Cloud path {} is used for running the job: {}', main_jar_path, job_name) + else: + logger.info('Uploading jar from {} to cloud for running job: {}', + main_jar_path, job_name) + main_jar_cloud_path = self._datalake.upload_file_to_workdir(main_jar_path) + logger.info('{} is uploaded to {} for running job: {}', + main_jar_path, main_jar_cloud_path, job_name) reference_file_paths = [] for file_path in reference_files_path: @@ -263,8 +276,13 @@ def create_spark_batch_job(self, job_name, main_file, class_name=None, executor_cores = self.EXECUTOR_SIZE[self._executor_size]['Cores'] executor_memory = self.EXECUTOR_SIZE[self._executor_size]['Memory'] - # need to put the jar in as dependencies for pyspark job - jars = jars + [main_file] + # If we have a main jar, it needs to be added as dependencies for pyspark job + # Otherwise it's a PySpark job with Feathr JAR from Maven + if main_file: + if not python_files: + # These 2 parameters should not be empty at the same time + raise ValueError("Main JAR is not set for the Spark job") + jars = jars + [main_file] # If file=main_file, then it's using only Scala Spark # If file=python_files[0], then it's using Pyspark diff --git a/feathr_project/feathr/spark_provider/noop-1.0.jar b/feathr_project/feathr/spark_provider/noop-1.0.jar index 24f7cb77584e78fb2be1ad89c6464a28fb324ce2..6b3b9ba56a34bf49524765c2fac4f4e21115f793 100644 GIT binary patch delta 1845 zcmX@XyH z@ZaF#{L&n~d;8RguUoEX&Eh$DWuxxxg&X$QrO#ntjCv~Ya_`5_PtKlRb8rQ3o5sII z{HJDJ&C%<&ss8vi?Cr()2j;!)c4z!t=I#CF<^HiM=IkroSoUyZr7DQ4Hf&> zLP4Q-SE$VJo5<(5`@!kdLPyCaTbU(B)@zg-dA8anCY_67b~wH&$umdQ^SNbW&H9Y$ z(8m{M7<`jm<1Mq-F`KpRgnX6S+-0i(mm6s&F-Haw&Z>{l@w>TR5 z{L6NhCCn$EN3ZIC_3VsAMXFL&x~YqodZb;w#l|IvSxwtNp02vf5_qb0k5%mLbCJzz zTn5LEsR((bdY)g|^IY`AwI~xUiw@K8ha2l>ywWgV(7n>#ZWZr^DeRv;+IRL|(y%hP zkaS!;T$5??&)kE_%2W0x?}+SZDp@7y?EP7m-23x`uB!8U|!(Ajn|$=&il)bNOj_QCqux= z4@$5A>HMN}eYlExV8TRI!UI*3T9KGrkdq2m$^ll%9oSy>6{w<`k%2)BO{HIcet}+c zPGWI!uD`7}v!g)m#yd|M%i%N_5voJxnoq1Pl zE{mLsJH03^ZOZJRYu=e#XYW6;VfW?lahsZ<@=I2v zMt8cVmqz>kRn62s*Y|%S=gmVoIV<(H?w%B}t9SXitL_~g27hY%R=!saKk7AAmha-@ z`d!zVcglR9p?=V#=i*1>7bj11e!nr9aodjX_Pibw4VT^c*26u&!1ArX!7o9THwo7t zAK2X{4-ZvfP)&ZusLceVgqh4iln;}QkPI-Sr*vB0-wX^pYgPsZb%x3JS(PUHv$5AN z4LP0n$Ux-oXVJiKHyn1am{2IL+#9m8f8wuz?dys;*9qO;B%APW-!Z=+c8;lOH=f=p ze)srJ_KCOq)@{<2(4G?WwAkdd@q`#rr)jJIe0}=;i`xFB!i#!;vAi=nKdo$P{+!F+ zJ=xn>J7=WTFni1t`d?Xf;CjZh%|4sniL0F6E4sHnXZ!7icS8kq(l%Sx)$gePCYfa> zddW)7a!HZ)G&7g?qUm~DekT9S>Djp5m1TKj`=4?@XTzY?OslrPVBrmHV!xsC$!oDx zZFs=Y|k3;W-QwFIxFMZ`Zn=$oPIMn?8egwY0~ameo<+ zTs)gr8Opritv~pjd9l-RAI^6@8+eb172I-e;kux}T`zrjU)7KI<`qf7yBi_`!+l?V zuXz%=-g`|~_pg*RDbwY4-&UPFq`k$V`^d$44gJ!-6H-p~yM=h>T{5c<(Ou;Hp}y_k z_k+!ow0DZ7CLelj(YyWTPq&Z6GmVB5g^N;JL_~f(9yFroL z#bSz9vQIwCVpPw-fGzt0GYt?3FuZkK39JMlnURr6gc(*%VCF`Ij?+*bh`b5Yi6wWM zLiB;jC5>Ovv?4NVfH%w#P%SmtnpK8Lj%9Krt1QTC6+oUh3oxC^@}a3a2$X37$>=Fy qCTxV&OBxS>6oHbtA=554Fq4~wPb|Qjl?|kd3kcT%-ImD?;sF5MYw^hd delta 999 zcmZ20a)Os9z?+$civa{!6ejYh@-Zu%45@hebZ0J5q=jLky;S{Y-!rFuymj?1@_OrP zojY@WbCAIm;|EWRbiB@;(DBxF;$VNWBy5V9_5qcOopZd5i@i$bYONKSWE}Ka%d2wd zobDiD@1WoVNqwxH4vDPyk8HYP!g4*e&3C_MC!w{n8cF4{0sme|AG<(D^+R z&-dK)i_CAi)bjbFR(!Yk#H&u>H}`Dx3J>0|mvi*zL;G|mD|74UQ`JwW32(P3T*>FO zW#=m%r~k`;1guHD`G86HUw^gkgwsZDYmCy8^5Wk*uKy=v>~>E@#;}Z!QN?4XuuS#w zsmJY__aA)_KCz%_-Sc?@{$KaQg9Q{zleJj1nSfL{i#dpzz+xlF0F1OL9g^R;fWf5C z#K53Fc|MEsBv$tN*1+iO!wM3$=ht)WbDx~cuNv{&OkzvU2JULn&$^{YxU(kz`4gGz z&87SKZ1K-?&nM=knNCr*(C>{~%yxYRi>{A)>3+BQ^5?H|1XL-HGZk$2DCm?M_}g7Z;av+2r9%LrJ+;yXqD1?|xf7XF-MP zR{j|7h+BJeYLm9ReVB0VkNLhm%V*S1KdAMxV2|>&Lo1AAdE$3+PK)qbw?XN3RXX3d zrOAadZ8Np48C25^ymxSaYVQ_c% z`{Z5hMmC6)hn(<$X%7sRG#&%eh-4Vx&B!Fe49h^17jVch-C~@)k3$w@!(AYcooVs{ z7TL)XoIDCFz*vGB1Ev(<907*^{~bUIRkUEnaWG&+DqJ}O!;;22ka8tp98ccOsl{}b Y8Ny?k%*!Rj$HBnCU From 328cce5a4a3fdad3476bc470495f85cc92409d85 Mon Sep 17 00:00:00 2001 From: Chen Xu Date: Sat, 11 Jun 2022 01:48:34 +0800 Subject: [PATCH 3/4] Print Maven package artifact id when using it --- .../feathr/spark_provider/_databricks_submission.py | 4 ++-- feathr_project/feathr/spark_provider/_synapse_submission.py | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/feathr_project/feathr/spark_provider/_databricks_submission.py b/feathr_project/feathr/spark_provider/_databricks_submission.py index 114e0728a..b5368c4f3 100644 --- a/feathr_project/feathr/spark_provider/_databricks_submission.py +++ b/feathr_project/feathr/spark_provider/_databricks_submission.py @@ -143,8 +143,8 @@ def submit_feathr_job(self, job_name: str, main_jar_path: str, main_class_name: submission_params['new_cluster']['spark_conf'] = configuration submission_params['new_cluster']['custom_tags'] = job_tags # the feathr main jar file is anyway needed regardless it's pyspark or scala spark - if main_jar_path is None or main_jar_path=="": - logger.info("Main JAR file is not set, using default package from Maven") + if not main_jar_path: + logger.info(f"Main JAR file is not set, using default package '{FEATHR_MAVEN_ARTIFACT}' from Maven") submission_params['libraries'][0]['maven'] = { "coordinates": FEATHR_MAVEN_ARTIFACT } else: submission_params['libraries'][0]['jar'] = self.upload_or_get_cloud_path(main_jar_path) diff --git a/feathr_project/feathr/spark_provider/_synapse_submission.py b/feathr_project/feathr/spark_provider/_synapse_submission.py index 77d836e8f..6476c3d77 100644 --- a/feathr_project/feathr/spark_provider/_synapse_submission.py +++ b/feathr_project/feathr/spark_provider/_synapse_submission.py @@ -109,7 +109,7 @@ def submit_feathr_job(self, job_name: str, main_jar_path: str = None, main_clas # This is a JAR job # Azure Synapse/Livy doesn't allow JAR job starts from Maven directly, we must have a jar file uploaded. # so we have to use a dummy jar as the main file. - logger.info("Main JAR file is not set, using default package from Maven") + logger.info(f"Main JAR file is not set, using default package '{FEATHR_MAVEN_ARTIFACT}' from Maven") # Add Maven dependency to the job if "spark.jars.packages" in cfg: cfg["spark.jars.packages"] = ",".join( From b9d35e85d3810918d9811f4d37ae7ed54c2e0a92 Mon Sep 17 00:00:00 2001 From: Chen Xu Date: Sat, 11 Jun 2022 03:14:26 +0800 Subject: [PATCH 4/4] Add e2e test for Maven job submission --- .../spark_provider/_synapse_submission.py | 30 +++-- .../test/test_azure_spark_maven_e2e.py | 63 ++++++++++ .../feathr_config_maven.yaml | 118 ++++++++++++++++++ 3 files changed, 200 insertions(+), 11 deletions(-) create mode 100644 feathr_project/test/test_azure_spark_maven_e2e.py create mode 100644 feathr_project/test/test_user_workspace/feathr_config_maven.yaml diff --git a/feathr_project/feathr/spark_provider/_synapse_submission.py b/feathr_project/feathr/spark_provider/_synapse_submission.py index 6476c3d77..98d4d05fd 100644 --- a/feathr_project/feathr/spark_provider/_synapse_submission.py +++ b/feathr_project/feathr/spark_provider/_synapse_submission.py @@ -102,26 +102,30 @@ def submit_feathr_job(self, job_name: str, main_jar_path: str = None, main_clas configuration (Dict[str, str]): Additional configs for the spark job """ - cfg = configuration.copy() # We don't want to mess up input parameters + if configuration: + cfg = configuration.copy() # We don't want to mess up input parameters + else: + cfg = {} if not main_jar_path: # We don't have the main jar, use Maven + # Add Maven dependency to the job configuration + if "spark.jars.packages" in cfg: + cfg["spark.jars.packages"] = ",".join( + [cfg["spark.jars.packages"], FEATHR_MAVEN_ARTIFACT]) + else: + cfg["spark.jars.packages"] = FEATHR_MAVEN_ARTIFACT + if not python_files: # This is a JAR job # Azure Synapse/Livy doesn't allow JAR job starts from Maven directly, we must have a jar file uploaded. # so we have to use a dummy jar as the main file. logger.info(f"Main JAR file is not set, using default package '{FEATHR_MAVEN_ARTIFACT}' from Maven") - # Add Maven dependency to the job - if "spark.jars.packages" in cfg: - cfg["spark.jars.packages"] = ",".join( - [cfg["spark.jars.packages"], FEATHR_MAVEN_ARTIFACT]) - else: - cfg["spark.jars.packages"] = FEATHR_MAVEN_ARTIFACT # Use the no-op jar as the main file # This is a dummy jar which contains only one `org.example.Noop` class with one empty `main` function which does nothing current_dir = pathlib.Path(__file__).parent.resolve() main_jar_path = os.path.join(current_dir, "noop-1.0.jar") else: - # This is a PySpark job + # This is a PySpark job, no more things to do pass main_jar_cloud_path = None if main_jar_path: @@ -136,6 +140,10 @@ def submit_feathr_job(self, job_name: str, main_jar_path: str = None, main_clas main_jar_cloud_path = self._datalake.upload_file_to_workdir(main_jar_path) logger.info('{} is uploaded to {} for running job: {}', main_jar_path, main_jar_cloud_path, job_name) + else: + # We don't have the main Jar, and this is a PySpark job so we don't use `noop.jar` either + # Keep `main_jar_cloud_path` as `None` as we already added maven package into cfg + pass reference_file_paths = [] for file_path in reference_files_path: @@ -279,10 +287,10 @@ def create_spark_batch_job(self, job_name, main_file, class_name=None, # If we have a main jar, it needs to be added as dependencies for pyspark job # Otherwise it's a PySpark job with Feathr JAR from Maven if main_file: - if not python_files: - # These 2 parameters should not be empty at the same time - raise ValueError("Main JAR is not set for the Spark job") jars = jars + [main_file] + elif not python_files: + # These 2 parameters should not be empty at the same time + raise ValueError("Main JAR is not set for the Spark job") # If file=main_file, then it's using only Scala Spark # If file=python_files[0], then it's using Pyspark diff --git a/feathr_project/test/test_azure_spark_maven_e2e.py b/feathr_project/test/test_azure_spark_maven_e2e.py new file mode 100644 index 000000000..5aa51b4ab --- /dev/null +++ b/feathr_project/test/test_azure_spark_maven_e2e.py @@ -0,0 +1,63 @@ +import os +from datetime import datetime, timedelta +from pathlib import Path + +from click.testing import CliRunner +from feathr import BOOLEAN, FLOAT, INT32, ValueType +from feathr import FeathrClient +from feathr import ValueType +from feathr.utils.job_utils import get_result_df +from feathr import (BackfillTime, MaterializationSettings) +from feathr import FeatureQuery +from feathr import ObservationSettings +from feathr import RedisSink, HdfsSink +from feathr import TypedKey +from feathrcli.cli import init +import pytest + +from test_fixture import (basic_test_setup, get_online_test_table_name) + +def test_feathr_online_store_agg_features(): + """ + Test FeathrClient() get_online_features and batch_get can get data correctly. + """ + + online_test_table = get_online_test_table_name("nycTaxiCITable") + test_workspace_dir = Path( + __file__).parent.resolve() / "test_user_workspace" + # os.chdir(test_workspace_dir) + + # The `feathr_runtime_location` was commented out in this config file, so feathr should use + # Maven package as the dependency and `noop.jar` as the main file + client = basic_test_setup(os.path.join(test_workspace_dir, "feathr_config_maven.yaml")) + + backfill_time = BackfillTime(start=datetime( + 2020, 5, 20), end=datetime(2020, 5, 20), step=timedelta(days=1)) + redisSink = RedisSink(table_name=online_test_table) + settings = MaterializationSettings("nycTaxiTable", + sinks=[redisSink], + feature_names=[ + "f_location_avg_fare", "f_location_max_fare"], + backfill_time=backfill_time) + client.materialize_features(settings) + # just assume the job is successful without validating the actual result in Redis. Might need to consolidate + # this part with the test_feathr_online_store test case + client.wait_job_to_finish(timeout_sec=900) + + res = client.get_online_features(online_test_table, '265', [ + 'f_location_avg_fare', 'f_location_max_fare']) + # just assme there are values. We don't hard code the values for now for testing + # the correctness of the feature generation should be garunteed by feathr runtime. + # ID 239 and 265 are available in the `DOLocationID` column in this file: + # https://s3.amazonaws.com/nyc-tlc/trip+data/green_tripdata_2020-04.csv + # View more detials on this dataset: https://www1.nyc.gov/site/tlc/about/tlc-trip-record-data.page + assert len(res) == 2 + assert res[0] != None + assert res[1] != None + res = client.multi_get_online_features(online_test_table, + ['239', '265'], + ['f_location_avg_fare', 'f_location_max_fare']) + assert res['239'][0] != None + assert res['239'][1] != None + assert res['265'][0] != None + assert res['265'][1] != None \ No newline at end of file diff --git a/feathr_project/test/test_user_workspace/feathr_config_maven.yaml b/feathr_project/test/test_user_workspace/feathr_config_maven.yaml new file mode 100644 index 000000000..ed3af5826 --- /dev/null +++ b/feathr_project/test/test_user_workspace/feathr_config_maven.yaml @@ -0,0 +1,118 @@ +# DO NOT MOVE OR DELETE THIS FILE + +# This file contains the configurations that are used by Feathr +# All the configurations can be overwritten by environment variables with concatenation of `__` for different layers of this config file. +# For example, `feathr_runtime_location` for databricks can be overwritten by setting this environment variable: +# SPARK_CONFIG__DATABRICKS__FEATHR_RUNTIME_LOCATION +# Another example would be overwriting Redis host with this config: `ONLINE_STORE__REDIS__HOST` +# For example if you want to override this setting in a shell environment: +# export ONLINE_STORE__REDIS__HOST=feathrazure.redis.cache.windows.net + +# version of API settings +api_version: 1 +project_config: + project_name: 'project_feathr_integration_test' + # Information that are required to be set via environment variables. + required_environment_variables: + # the environemnt variables are required to run Feathr + # Redis password for your online store + - 'REDIS_PASSWORD' + # client IDs and client Secret for the service principal. Read the getting started docs on how to get those information. + - 'AZURE_CLIENT_ID' + - 'AZURE_TENANT_ID' + - 'AZURE_CLIENT_SECRET' + optional_environment_variables: + # the environemnt variables are optional, however you will need them if you want to use some of the services: + - ADLS_ACCOUNT + - ADLS_KEY + - WASB_ACCOUNT + - WASB_KEY + - S3_ACCESS_KEY + - S3_SECRET_KEY + - JDBC_TABLE + - JDBC_USER + - JDBC_PASSWORD + - KAFKA_SASL_JAAS_CONFIG + +offline_store: + # paths starts with abfss:// or abfs:// + # ADLS_ACCOUNT and ADLS_KEY should be set in environment variable if this is set to true + adls: + adls_enabled: true + + # paths starts with wasb:// or wasbs:// + # WASB_ACCOUNT and WASB_KEY should be set in environment variable + wasb: + wasb_enabled: true + + # paths starts with s3a:// + # S3_ACCESS_KEY and S3_SECRET_KEY should be set in environment variable + s3: + s3_enabled: true + # S3 endpoint. If you use S3 endpoint, then you need to provide access key and secret key in the environment variable as well. + s3_endpoint: 's3.amazonaws.com' + + # jdbc endpoint + jdbc: + jdbc_enabled: true + jdbc_database: 'feathrtestdb' + jdbc_table: 'feathrtesttable' + + # snowflake endpoint + snowflake: + url: "dqllago-ol19457.snowflakecomputing.com" + user: "feathrintegration" + role: "ACCOUNTADMIN" + +spark_config: + # choice for spark runtime. Currently support: azure_synapse, databricks + # The `databricks` configs will be ignored if `azure_synapse` is set and vice versa. + spark_cluster: 'azure_synapse' + # configure number of parts for the spark output for feature generation job + spark_result_output_parts: '1' + + azure_synapse: + dev_url: 'https://feathrazuretest3synapse.dev.azuresynapse.net' + pool_name: 'spark3' + # workspace dir for storing all the required configuration files and the jar resources + workspace_dir: 'abfss://feathrazuretest3fs@feathrazuretest3storage.dfs.core.windows.net/feathr_test_workspace' + executor_size: 'Small' + executor_num: 4 + + # Feathr Job configuration. Support local paths, path start with http(s)://, and paths start with abfs(s):// + # this is the default location so end users don't have to compile the runtime again. + # feathr_runtime_location: wasbs://public@azurefeathrstorage.blob.core.windows.net/feathr-assembly-0.5.0-SNAPSHOT.jar + # Unset this value will use default package on Maven + # feathr_runtime_location: "../../target/scala-2.12/feathr-assembly-0.5.0.jar" + databricks: + # workspace instance + workspace_instance_url: 'https://adb-5638037984879289.9.azuredatabricks.net/' + workspace_token_value: '' + # config string including run time information, spark version, machine size, etc. + # the config follows the format in the databricks documentation: https://docs.microsoft.com/en-us/azure/databricks/dev-tools/api/2.0/jobs + config_template: {'run_name':'','new_cluster':{'spark_version':'9.1.x-scala2.12','node_type_id':'Standard_F4s','num_workers':2,'spark_conf':{}},'libraries':[{'jar':''}],'spark_jar_task':{'main_class_name':'','parameters':['']}} + # Feathr Job location. Support local paths, path start with http(s)://, and paths start with dbfs:/ + work_dir: 'dbfs:/feathr_getting_started' + + # this is the default location so end users don't have to compile the runtime again. + # Unset this value will use default package on Maven + # feathr_runtime_location: "../../target/scala-2.12/feathr-assembly-0.5.0.jar" + +online_store: + redis: + # Redis configs to access Redis cluster + host: 'feathrazuretest3redis.redis.cache.windows.net' + port: 6380 + ssl_enabled: True + +feature_registry: + purview: + # Registry configs + # register type system in purview during feathr client initialization. This is only required to be executed once. + type_system_initialization: true + # configure the name of the purview endpoint + purview_name: 'feathrazuretest3-purview1' + # delimiter indicates that how the project/workspace name, feature names etc. are delimited. By default it will be '__' + # this is for global reference (mainly for feature sharing). For exmaple, when we setup a project called foo, and we have an anchor called 'taxi_driver' and the feature name is called 'f_daily_trips' + # the feature will have a globally unique name called 'foo__taxi_driver__f_daily_trips' + delimiter: '__' \ No newline at end of file