瀏覽代碼

Oozie MapReduce action should be able to accept s3a and abfs as Jar name (#2974)

Jira: CDPD-44349
Mahesh Balakrishnan 3 年之前
父節點
當前提交
e4f14839e8
共有 1 個文件被更改,包括 9 次插入3 次删除
  1. 9 3
      desktop/libs/liboozie/src/liboozie/submission2.py

+ 9 - 3
desktop/libs/liboozie/src/liboozie/submission2.py

@@ -557,7 +557,10 @@ STORED AS TEXTFILE %s""" % (self.properties.get('send_result_path'),
         jar_path = node.data['properties'].get('jar_path')
         if jar_path:
           if not jar_path.startswith('/'):  # If workspace relative path
-            jar_path = self.fs.join(self.job.deployment_dir, jar_path)
+            if jar_path.startswith('s3a://') or jar_path.startswith('abfs://'):
+              jar_path = jar_path
+            else:
+              jar_path = self.fs.join(self.job.deployment_dir, jar_path)
           if not jar_path.startswith(lib_path):  # If not already in lib
             files.append(jar_path)
 
@@ -574,12 +577,15 @@ STORED AS TEXTFILE %s""" % (self.properties.get('send_result_path'),
       if files:
         for jar_file in files:
           LOG.debug("Updating %s" % jar_file)
-          jar_lib_path = self.fs.join(lib_path, self.fs.basename(jar_file))
+          if jar_file.startswith('s3a://') or jar_file.startswith('abfs://'):
+            jar_lib_path = jar_file
+          else:
+            jar_lib_path = self.fs.join(lib_path, self.fs.basename(jar_file))
           # Refresh if needed
           if self.fs.exists(jar_lib_path) and self.fs.exists(jar_file):
             stat_src = self.fs.stats(jar_file)
             stat_dest = self.fs.stats(jar_lib_path)
-            if stat_src.fileId != stat_dest.fileId:
+            if hasattr(stat_src, 'fileId') and hasattr(stat_dest, 'fileId') and stat_src.fileId != stat_dest.fileId:
               self.fs.remove(jar_lib_path, skip_trash=True)
           self.fs.copyfile(jar_file, jar_lib_path)