docs: def readtb(paths, num_procs=1, java_mem_mb=256, ignore_logs=True):
      """Read typedbytes sequence files on HDFS (with optional compression).
  
      By default, ignores files who's names start with an underscore '_' as they
      are log files.  This allows you to cat a directory that may be a variety of
      outputs from hadoop (e.g., _SUCCESS, _logs).  This works on directories and
      files.  The KV pairs may be interleaved between files
      (they are read in parallel).
  
      :param paths: HDFS path (str) or paths (iterator)
      :param num_procs: Number of reading procs to open (default 1)
      :param java_mem_mb: Integer of java heap size in MB (default 256)
      :param ignore_logs: If True, ignore all files who's name starts with an underscore.  Defaults to True.
      :returns: An iterator of key, value pairs.
      :raises: IOError: An error occurred reading the directory (e.g., not available).
      """
      import select
      hstreaming = _find_hstreaming()
      if isinstance(paths, (str, unicode)):
          paths = [paths]
      read_fds = set()
      procs = {}
      tb_fps = {}
  
      def _open_tb(cur_path):
          cmd = 'hadoop jar %s dumptb %s' % (hstreaming, cur_path)
          read_fd, write_fd = os.pipe()
          write_fp = os.fdopen(write_fd, 'w')
          p = _hadoop_fs_command(cmd, stdout=write_fp, java_mem_mb=java_mem_mb)
          write_fp.close()
          read_fds.add(read_fd)
          procs[read_fd] = p
          tb_fps[read_fd] = hadoopy.TypedBytesFile(read_fd=read_fd)
  
      def _path_gen():
          for root_path in paths:
              try:
                  all_paths = ls(root_path)
              except IOError:
                  raise IOError("No such file or directory: '%s'" % root_path)
              if ignore_logs:
                  # Ignore any files that start with an underscore
                  keep_file = lambda x: os.path.basename(x)[0] != '_'
                  all_paths = filter(keep_file, all_paths)
              for cur_path in all_paths:
                  yield _open_tb(cur_path)
      try:
          path_gen = _path_gen()
          for x in range(num_procs):
              try:
                  path_gen.next()
              except (AttributeError, StopIteration):
                  path_gen = None
          while read_fds:
              cur_fds = select.select(read_fds, [], [])[0]
              for read_fd in cur_fds:
                  p = procs[read_fd]
                  tp_fp = tb_fps[read_fd]
                  try:
                      yield tp_fp.next()
                  except StopIteration:
                      p.wait()
                      del procs[read_fd]
                      del tb_fps[read_fd]
                      del p
                      os.close(read_fd)
                      read_fds.remove(read_fd)
                      try:
                          path_gen.next()
                      except (AttributeError, StopIteration):
                          path_gen = None
      finally:
          # Cleanup outstanding procs
          for p in procs.values():
              p.kill()
              p.wait()