<commit_before># BEGIN_COPYRIGHT
# 
# Copyright 2012 CRS4.
# 
# Licensed under the Apache License, Version 2.0 (the "License"); you may not
# use this file except in compliance with the License. You may obtain a copy
# of the License at
# 
#   http://www.apache.org/licenses/LICENSE-2.0
# 
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
# 
# END_COPYRIGHT

"""
The hadut module provides access to some Hadoop functionalities
available via the Hadoop shell.
"""

import os, subprocess

import pydoop
import pydoop.utils as utils
import pydoop.hadoop_utils as hu
import pydoop.hdfs as hdfs


#--- FIXME: perhaps we need a more sophisticated tool for setting args ---
GENERIC_ARGS = frozenset([
  "-conf", "-D", "-fs", "-jt", "-files", "-libjars", "-archives"
  ])

CSV_ARGS = frozenset([
  "-files", "-libjars", "-archives"
  ])

# generic args must go before command-specific args
def _pop_generic_args(args):
  generic_args = []
  i = len(args) - 1
  while i >= 0:
    if args[i] in GENERIC_ARGS:
      try:
        args[i+1]
      except IndexError:
        raise ValueError("option %s has no value" % args[i])
      generic_args.extend(args[i:i+2])
      del args[i:i+2]
    i -= 1
  return generic_args

# -files f1 -files f2 --> -files f1,f2
def _merge_csv_args(args):
  merge_map = {}
  i = len(args) - 1
  while i >= 0:
    if args[i] in CSV_ARGS:
      try:
        args[i+1]
      except IndexError:
        raise ValueError("option %s has no value" % args[i])
      k, v = args[i:i+2]
      merge_map.setdefault(k, []).append(v.strip())
      del args[i:i+2]
    i -= 1
  for k, vlist in merge_map.iteritems():
    args.extend([k, ",".join(vlist)])

# FIXME: the above functions share a lot of code
#-------------------------------------------------------------------------


def _construct_property_args(prop_dict):
  return sum((['-D', '%s=%s' % p] for p in prop_dict.iteritems()), [])


def run_cmd(cmd, args=None, properties=None, hadoop_home=None,
            hadoop_conf_dir=None, logger=None):
  """
  Run a Hadoop command.

  If the command succeeds, return its output; if it fails, raise a
  ``RuntimeError`` with its error output as the message.

  .. code-block:: python

    >>> import uuid
    >>> properties = {'dfs.block.size': 32*2**20}
    >>> args = ['-put', 'hadut.py', uuid.uuid4().hex]
    >>> res = run_cmd('fs', args, properties)
    >>> res
    ''
    >>> print run_cmd('dfsadmin', ['-help', 'report'])
    -report: Reports basic filesystem information and statistics.
    >>> try:
    ...     run_cmd('foo')
    ... except RuntimeError as e:
    ...     print e
    ... 
    Exception in thread "main" java.lang.NoClassDefFoundError: foo
    ...
  """
  if logger is None:
    logger = utils.NullLogger()
  hadoop = pydoop.hadoop_exec(hadoop_home=hadoop_home)
  _args = [hadoop]
  if hadoop_conf_dir:
    _args.extend(["--config", hadoop_conf_dir])
  _args.append(cmd)
  if properties:
    _args.extend(_construct_property_args(properties))
  if args:
    if isinstance(args, basestring):
      args = args.split()
    _merge_csv_args(args)
    gargs = _pop_generic_args(args)
    for seq in gargs, args:
      _args.extend(map(str, seq))
  logger.debug("cmd: %s" % " ".join(_args))
  p = subprocess.Popen(_args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  output, error = p.communicate()
  if p.returncode:
    raise RuntimeError(error)
  return output


def get_task_trackers(properties=None, hadoop_conf_dir=None):
  """
  Get the list of task trackers in the Hadoop cluster.

  Each element in the returned list is in the ``(host, port)`` format.
  ``properties`` is passed to :func:`run_cmd`.
  """
  stdout = run_cmd("job", ["-list-active-trackers"],
                   properties=properties, hadoop_conf_dir=hadoop_conf_dir)
  task_trackers = []
  for l in stdout.splitlines():<commit_msg>Use shlex.split instead of whitespace splitting for command line<commit_after># BEGIN_COPYRIGHT
# 
# Copyright 2012 CRS4.
# 
# Licensed under the Apache License, Version 2.0 (the "License"); you may not
# use this file except in compliance with the License. You may obtain a copy
# of the License at
# 
#   http://www.apache.org/licenses/LICENSE-2.0
# 
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
# 
# END_COPYRIGHT

"""
The hadut module provides access to some Hadoop functionalities
available via the Hadoop shell.
"""

import os, subprocess
import shlex

import pydoop
import pydoop.utils as utils
import pydoop.hadoop_utils as hu
import pydoop.hdfs as hdfs


#--- FIXME: perhaps we need a more sophisticated tool for setting args ---
GENERIC_ARGS = frozenset([
  "-conf", "-D", "-fs", "-jt", "-files", "-libjars", "-archives"
  ])

CSV_ARGS = frozenset([
  "-files", "-libjars", "-archives"
  ])

# generic args must go before command-specific args
def _pop_generic_args(args):
  generic_args = []
  i = len(args) - 1
  while i >= 0:
    if args[i] in GENERIC_ARGS:
      try:
        args[i+1]
      except IndexError:
        raise ValueError("option %s has no value" % args[i])
      generic_args.extend(args[i:i+2])
      del args[i:i+2]
    i -= 1
  return generic_args

# -files f1 -files f2 --> -files f1,f2
def _merge_csv_args(args):
  merge_map = {}
  i = len(args) - 1
  while i >= 0:
    if args[i] in CSV_ARGS:
      try:
        args[i+1]
      except IndexError:
        raise ValueError("option %s has no value" % args[i])
      k, v = args[i:i+2]
      merge_map.setdefault(k, []).append(v.strip())
      del args[i:i+2]
    i -= 1
  for k, vlist in merge_map.iteritems():
    args.extend([k, ",".join(vlist)])

# FIXME: the above functions share a lot of code
#-------------------------------------------------------------------------


def _construct_property_args(prop_dict):
  return sum((['-D', '%s=%s' % p] for p in prop_dict.iteritems()), [])


def run_cmd(cmd, args=None, properties=None, hadoop_home=None,
            hadoop_conf_dir=None, logger=None):
  """
  Run a Hadoop command.

  If the command succeeds, return its output; if it fails, raise a
  ``RuntimeError`` with its error output as the message.

  .. code-block:: python

    >>> import uuid
    >>> properties = {'dfs.block.size': 32*2**20}
    >>> args = ['-put', 'hadut.py', uuid.uuid4().hex]
    >>> res = run_cmd('fs', args, properties)
    >>> res
    ''
    >>> print run_cmd('dfsadmin', ['-help', 'report'])
    -report: Reports basic filesystem information and statistics.
    >>> try:
    ...     run_cmd('foo')
    ... except RuntimeError as e:
    ...     print e
    ... 
    Exception in thread "main" java.lang.NoClassDefFoundError: foo
    ...
  """
  if logger is None:
    logger = utils.NullLogger()
  hadoop = pydoop.hadoop_exec(hadoop_home=hadoop_home)
  _args = [hadoop]
  if hadoop_conf_dir:
    _args.extend(["--config", hadoop_conf_dir])
  _args.append(cmd)
  if properties:
    _args.extend(_construct_property_args(properties))
  if args:
    if isinstance(args, basestring):
      args = shlex.split(args)
    _merge_csv_args(args)
    gargs = _pop_generic_args(args)
    for seq in gargs, args:
      _args.extend(map(str, seq))
  logger.debug("cmd: %s" % " ".join(_args))
  p = subprocess.Popen(_args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  output, error = p.communicate()
  if p.returncode:
    raise RuntimeError(error)
  return output


def get_task_trackers(properties=None, hadoop_conf_dir=None):
  """
  Get the list of task trackers in the Hadoop cluster.

  Each element in the returned list is in the ``(host, port)`` format.
  ``properties`` is passed to :func:`run_cmd`.
  """
  stdout = run_cmd("job", ["-list-active-trackers"],
                   properties=properties, hadoop_conf_dir=hadoop_conf_dir)
  task_trackers = []
  for l in stdout.splitlines():
{'fraction_non_alphanumeric': 0.14675685469259572, 'fraction_numerical': 0.007488827153037806, 'mean_word_length': 3.0469208211143695, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 51, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6368077', 'n_tokens_mistral': 2758, 'n_tokens_neox': 2691, 'n_words': 821}