Browse Source

[hadoop] Add config for Yarn cluster, Mapred port, and `submit_to' param

The `yarn_clusters' is a new section. The plural is to be consistent with HDFS
and MR config. But there is no plan to work with multiple Yarn clusters.

The JT port needs to be configured, in order for Oozie to know how to connect
to it.

Added a new `submit_to' boolean param, to mark which cluster to run jobs on.
bc Wong 13 years ago
parent
commit
8df29f939e

+ 16 - 0
desktop/conf.dist/hue.ini

@@ -199,5 +199,21 @@
     [[[default]]]
     [[[default]]]
       # Enter the host on which you are running the Hadoop JobTracker
       # Enter the host on which you are running the Hadoop JobTracker
       jobtracker_host=localhost
       jobtracker_host=localhost
+      # The port where the JobTracker IPC listens on
+      jobtracker_port=8021
       # Thrift plug-in port for the JobTracker
       # Thrift plug-in port for the JobTracker
       ## thrift_port=9290
       ## thrift_port=9290
+      # Whether to submit jobs to this cluster
+      ## submit_to=False
+
+  # Configuration for Yarn
+  # ------------------------------------------------------------------------
+  [[yarn_clusters]]
+
+    [[[default]]]
+      # Enter the host on which you are running the ResourceManager
+      resourcemanager_host=localhost
+      # The port where the ResourceManager IPC listens on
+      resourcemanager_port=8032
+      # Whether to submit jobs to this cluster
+      ## submit_to=False

+ 27 - 1
desktop/libs/hadoop/src/hadoop/cluster.py

@@ -23,12 +23,14 @@ from hadoop import conf
 import os
 import os
 import logging
 import logging
 
 
+LOG = logging.getLogger(__name__)
+
 def _make_filesystem(identifier):
 def _make_filesystem(identifier):
   choice = os.getenv("FB_FS")
   choice = os.getenv("FB_FS")
   if choice == "testing":
   if choice == "testing":
     path = os.path.join(get_build_dir(), "fs")
     path = os.path.join(get_build_dir(), "fs")
     if not os.path.isdir(path):
     if not os.path.isdir(path):
-      logging.warning(
+      LOG.warning(
         ("Could not find fs directory: %s. Perhaps you need to run " +
         ("Could not find fs directory: %s. Perhaps you need to run " +
         "manage.py filebrowser_test_setup?") % path)
         "manage.py filebrowser_test_setup?") % path)
     return LocalSubFileSystem(path)
     return LocalSubFileSystem(path)
@@ -52,6 +54,15 @@ def get_hdfs(identifier="default"):
   get_all_hdfs()
   get_all_hdfs()
   return FS_CACHE[identifier]
   return FS_CACHE[identifier]
 
 
+def get_hdfs_url(identifier="default"):
+  """Returns the host:port for the given HDFS"""
+  try:
+    hdfs = conf.HDFS_CLUSTERS[identifier]
+  except KeyError:
+    LOG.error("Misconfiguration: No HDFS cluster called '%s'" % (identifier,))
+    return None
+  return "%s:%s" % (hdfs.NN_HOST.get(), hdfs.NN_HDFS_PORT.get())
+
 def get_all_hdfs():
 def get_all_hdfs():
   global FS_CACHE
   global FS_CACHE
   if FS_CACHE is not None:
   if FS_CACHE is not None:
@@ -89,6 +100,21 @@ def all_mrclusters():
     MR_CACHE[identifier] = _make_mrcluster(identifier)
     MR_CACHE[identifier] = _make_mrcluster(identifier)
   return MR_CACHE
   return MR_CACHE
 
 
+def get_cluster_for_job_submission():
+  """
+  Check the `submit_to' for each MR/Yarn cluster, and return the
+  host:port of first one that enables submission.
+  """
+  for name in conf.YARN_CLUSTERS.keys():
+    yarn = conf.YARN_CLUSTERS[name]
+    if yarn.SUBMIT_TO.get():
+      return "%s:%s" % (yarn.RM_HOST.get(), yarn.RM_PORT.get())
+  for name in conf.MR_CLUSTERS.keys():
+    mr = conf.MR_CLUSTERS[name]
+    if mr.SUBMIT_TO.get():
+      return "%s:%s" % (mr.JT_HOST.get(), mr.JT_PORT.get())
+  return None
+
 def clear_caches():
 def clear_caches():
   """
   """
   Clears cluster's internal caches.  Returns
   Clears cluster's internal caches.  Returns

+ 51 - 8
desktop/libs/hadoop/src/hadoop/conf.py

@@ -124,7 +124,7 @@ HDFS_CLUSTERS = UnspecifiedConfigSection(
   each=ConfigSection(
   each=ConfigSection(
     help="Information about a single HDFS cluster",
     help="Information about a single HDFS cluster",
     members=dict(
     members=dict(
-      NN_HOST=Config("namenode_host", help="IP for name node"),
+      NN_HOST=Config("namenode_host", help="Host/IP for name node"),
       NN_THRIFT_PORT=Config("thrift_port", help="Thrift port for name node", default=10090,
       NN_THRIFT_PORT=Config("thrift_port", help="Thrift port for name node", default=10090,
                             type=int),
                             type=int),
       NN_HDFS_PORT=Config("hdfs_port", help="Hadoop IPC port for the name node", default=8020,
       NN_HDFS_PORT=Config("hdfs_port", help="Hadoop IPC port for the name node", default=8020,
@@ -155,22 +155,42 @@ MR_CLUSTERS = UnspecifiedConfigSection(
   each=ConfigSection(
   each=ConfigSection(
     help="Information about a single MapReduce cluster",
     help="Information about a single MapReduce cluster",
     members=dict(
     members=dict(
-      JT_HOST=Config("jobtracker_host", help="IP for JobTracker"),
+      JT_HOST=Config("jobtracker_host", help="Host/IP for JobTracker"),
+      JT_PORT=Config("jobtracker_port",
+                     default=8021,
+                     help="Service port for the JobTracker",
+                     type=int),
       JT_THRIFT_PORT=Config("thrift_port", help="Thrift port for JobTracker", default=9290,
       JT_THRIFT_PORT=Config("thrift_port", help="Thrift port for JobTracker", default=9290,
                             type=int),
                             type=int),
       JT_KERBEROS_PRINCIPAL=Config("jt_kerberos_principal", help="Kerberos principal for JobTracker",
       JT_KERBEROS_PRINCIPAL=Config("jt_kerberos_principal", help="Kerberos principal for JobTracker",
                                    default="mapred", type=str),
                                    default="mapred", type=str),
       SECURITY_ENABLED=Config("security_enabled", help="Is running with Kerberos authentication",
       SECURITY_ENABLED=Config("security_enabled", help="Is running with Kerberos authentication",
-                              default=False, type=coerce_bool)
+                              default=False, type=coerce_bool),
+      SUBMIT_TO=Config('submit_to', help="Whether Hue should use this cluster to run jobs",
+                       default=False, type=coerce_bool),
     )
     )
   )
   )
 )
 )
 
 
-OOZIE_URL = Config(
-  key='oozie_url',
-  help='URL to Oozie server. This is required for job submission.',
-  default='http://localhost:11000/oozie',
-  type=str)
+YARN_CLUSTERS = UnspecifiedConfigSection(
+  "yarn_clusters",
+  help="One entry for each Yarn cluster. Currently only one cluster "
+       "(called 'default') is supported.",
+  each=ConfigSection(
+    help="Information about a single Yarn cluster",
+    members=dict(
+      RM_HOST=Config("resourcemanager_host",
+                     default='localhost',
+                     help="Host/IP for the ResourceManager"),
+      RM_PORT=Config("resourcemanager_port",
+                     default=8032,
+                     type=int,
+                     help="Service port for the ResourceManager"),
+      SUBMIT_TO=Config('submit_to', help="Whether Hue should use this cluster to run jobs",
+                       default=False, type=coerce_bool),
+    )
+  )
+)
 
 
 
 
 def config_validator():
 def config_validator():
@@ -192,14 +212,37 @@ def config_validator():
   res.extend(validate_path(HADOOP_EXAMPLES_JAR, is_dir=False))
   res.extend(validate_path(HADOOP_EXAMPLES_JAR, is_dir=False))
   res.extend(validate_path(HADOOP_STREAMING_JAR, is_dir=False))
   res.extend(validate_path(HADOOP_STREAMING_JAR, is_dir=False))
 
 
+  submit_to = [ ]
+
   # HDFS_CLUSTERS
   # HDFS_CLUSTERS
+  has_default = False
   for name in HDFS_CLUSTERS.keys():
   for name in HDFS_CLUSTERS.keys():
     cluster = HDFS_CLUSTERS[name]
     cluster = HDFS_CLUSTERS[name]
     res.extend(webhdfs.test_fs_configuration(cluster))
     res.extend(webhdfs.test_fs_configuration(cluster))
+    if name == 'default':
+      has_default = True
+  if not has_default:
+    res.append("hadoop.hdfs_clusters", "You should have an HDFS called 'default'.")
 
 
   # MR_CLUSTERS
   # MR_CLUSTERS
   for name in MR_CLUSTERS.keys():
   for name in MR_CLUSTERS.keys():
     cluster = MR_CLUSTERS[name]
     cluster = MR_CLUSTERS[name]
     res.extend(job_tracker.test_jt_configuration(cluster))
     res.extend(job_tracker.test_jt_configuration(cluster))
+    if cluster.SUBMIT_TO.get():
+      submit_to.append('mapred_clusters.' + name)
+
+  # Only one cluster should have submit_to
+  for name in YARN_CLUSTERS.keys():
+    cluster = YARN_CLUSTERS[name]
+    if cluster.SUBMIT_TO.get():
+      submit_to.append('yarn_clusters.' + name)
+
+  if len(submit_to) > 1:
+    res.append(("hadoop", "Only one cluster may enable 'submit_to'. "
+                "But it is enabled in the following clusters: " + 
+                ', '.join(submit_to)))
+  elif len(submit_to) == 0:
+    res.append(("hadoop", "Please designate one of the MapReduce or "
+                "Yarn clusters with `submit_to=true' in order to run jobs."))
 
 
   return res
   return res