docs: def run_local_analysis(output_dir, csv_file_pattern, schema, features):
    """Use pandas to analyze csv files.
  
    Produces a stats file and vocab files.
  
    Args:
      output_dir: output folder
      csv_file_pattern: list of csv file paths, may contain wildcards
      schema: CSV schema list
      features: features config
  
    Raises:
      ValueError: on unknown transfrorms/schemas
    """
    sys.stdout.write('Expanding any file patterns...\n')
    sys.stdout.flush()
    header = [column['name'] for column in schema]
    input_files = []
    for file_pattern in csv_file_pattern:
      input_files.extend(file_io.get_matching_files(file_pattern))
    sys.stdout.write('file list computed.\n')
    sys.stdout.flush()
  
    expand_defaults(schema, features)  # features are updated.
    inverted_features = invert_features(features)
    check_schema_transforms_match(schema, inverted_features)
  
    # Make a copy of inverted_features and update the target transform to be
    # identity or one hot depending on the schema.
    inverted_features_target = copy.deepcopy(inverted_features)
    for name, transforms in six.iteritems(inverted_features_target):
      transform_set = {x['transform'] for x in transforms}
      if transform_set == set([constant.TARGET_TRANSFORM]):
        target_schema = next(col['type'].lower() for col in schema if col['name'] == name)
        if target_schema in constant.NUMERIC_SCHEMA:
          inverted_features_target[name] = [{'transform': constant.IDENTITY_TRANSFORM}]
        else:
          inverted_features_target[name] = [{'transform': constant.ONE_HOT_TRANSFORM}]
  
    # initialize the results
    def _init_numerical_results():
      return {'min': float('inf'),
              'max': float('-inf'),
              'count': 0,
              'sum': 0.0}
    numerical_results = collections.defaultdict(_init_numerical_results)
    vocabs = collections.defaultdict(lambda: collections.defaultdict(int))
  
    num_examples = 0
    # for each file, update the numerical stats from that file, and update the set
    # of unique labels.
    for input_file in input_files:
      sys.stdout.write('Analyzing file %s...\n' % input_file)
      sys.stdout.flush()
      with file_io.FileIO(input_file, 'r') as f:
        for line in csv.reader(f):
          if len(header) != len(line):
            raise ValueError('Schema has %d columns but a csv line only has %d columns.' %
                             (len(header), len(line)))
          parsed_line = dict(zip(header, line))
          num_examples += 1
  
          for col_name, transform_set in six.iteritems(inverted_features_target):
            # All transforms in transform_set require the same analysis. So look
            # at the first transform.
            transform = next(iter(transform_set))
            if transform['transform'] in constant.TEXT_TRANSFORMS:
              separator = transform.get('separator', ' ')
              split_strings = parsed_line[col_name].split(separator)
  
              # If a label is in the row N times, increase it's vocab count by 1.
              # This is needed for TFIDF, but it's also an interesting stat.
              for one_label in set(split_strings):
                # Filter out empty strings
                if one_label:
                  vocabs[col_name][one_label] += 1
            elif transform['transform'] in constant.CATEGORICAL_TRANSFORMS:
              if parsed_line[col_name]:
                vocabs[col_name][parsed_line[col_name]] += 1
            elif transform['transform'] in constant.NUMERIC_TRANSFORMS:
              if not parsed_line[col_name].strip():
                continue
  
              numerical_results[col_name]['min'] = (
                min(numerical_results[col_name]['min'],
                    float(parsed_line[col_name])))
              numerical_results[col_name]['max'] = (
                max(numerical_results[col_name]['max'],
                    float(parsed_line[col_name])))
              numerical_results[col_name]['count'] += 1
              numerical_results[col_name]['sum'] += float(parsed_line[col_name])
  
      sys.stdout.write('file %s analyzed.\n' % input_file)
      sys.stdout.flush()
  
    # Write the vocab files. Each label is on its own line.
    vocab_sizes = {}
    for name, label_count in six.iteritems(vocabs):
      # df is now:
      # label1,count
      # label2,count
      # ...
      # where label1 is the most frequent label, and label2 is the 2nd most, etc.
      df = pd.DataFrame([{'label': label, 'count': count}
                         for label, count in sorted(six.iteritems(label_count),
                                                    key=lambda x: x[1],
                                                    reverse=True)],
                        columns=['label', 'count'])
      csv_string = df.to_csv(index=False, header=False)
  
      file_io.write_string_to_file(
          os.path.join(output_dir, constant.VOCAB_ANALYSIS_FILE % name),
          csv_string)
  
      vocab_sizes[name] = {'vocab_size': len(label_count)}
  
    # Update numerical_results to just have min/min/mean
    for col_name in numerical_results:
      if float(numerical_results[col_name]['count']) == 0:
        raise ValueError('Column %s has a zero count' % col_name)
      mean = (numerical_results[col_name]['sum'] /
              float(numerical_results[col_name]['count']))
      del numerical_results[col_name]['sum']
      del numerical_results[col_name]['count']
      numerical_results[col_name]['mean'] = mean
  
    # Write the stats file.
    numerical_results.update(vocab_sizes)
    stats = {'column_stats': numerical_results, 'num_examples': num_examples}
    file_io.write_string_to_file(
        os.path.join(output_dir, constant.STATS_FILE),
        json.dumps(stats, indent=2, separators=(',', ': ')))
  
    save_schema_features(schema, features, output_dir)