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)
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