docs: def run_cloud_analysis(output_dir, csv_file_pattern, bigquery_table, schema,
                         features):
    """Use BigQuery to analyze input date.
  
    Only one of csv_file_pattern or bigquery_table should be non-None.
  
    Args:
      output_dir: output folder
      csv_file_pattern: list of csv file paths, may contain wildcards
      bigquery_table: project_id.dataset_name.table_name
      schema: schema list
      features: features config
    """
  
    def _execute_sql(sql, table):
      """Runs a BigQuery job and dowloads the results into local memeory.
  
      Args:
        sql: a SQL string
        table: bq.ExternalDataSource or bq.Table
  
      Returns:
        A Pandas dataframe.
      """
      import google.datalab.bigquery as bq
      if isinstance(table, bq.ExternalDataSource):
        query = bq.Query(sql, data_sources={'csv_table': table})
      else:
        query = bq.Query(sql)
      return query.execute().result().to_dataframe()
  
    feature_analysis.expand_defaults(schema, features)  # features are updated.
    inverted_features = feature_analysis.invert_features(features)
    feature_analysis.check_schema_transforms_match(schema, inverted_features)
  
    import google.datalab.bigquery as bq
    if bigquery_table:
      table_name = '`%s`' % bigquery_table
      table = None
    else:
      table_name = 'csv_table'
      table = bq.ExternalDataSource(
          source=csv_file_pattern,
          schema=bq.Schema(schema))
  
    # 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}]
  
    numerical_vocab_stats = {}
    for col_name, transform_set in six.iteritems(inverted_features_target):
      sys.stdout.write('Analyzing column %s...\n' % col_name)
      sys.stdout.flush()
      # 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.CATEGORICAL_TRANSFORMS or
         transform['transform'] in constant.TEXT_TRANSFORMS):
        if transform['transform'] in constant.TEXT_TRANSFORMS:
          # Split strings on space, then extract labels and how many rows each
          # token is in. This is done by making two temp tables:
          #   SplitTable: each text row is made into an array of strings. The
          #       array may contain repeat tokens
          #   TokenTable: SplitTable with repeated tokens removed per row.
          # Then to flatten the arrays, TokenTable has to be joined with itself.
          # See the sections 'Flattening Arrays' and 'Filtering Arrays' at
          # https://cloud.google.com/bigquery/docs/reference/standard-sql/arrays
          separator = transform.get('separator', ' ')
          sql = ('WITH SplitTable AS '
                 '         (SELECT SPLIT({name}, \'{separator}\') as token_array FROM {table}), '
                 '     TokenTable AS '
                 '         (SELECT ARRAY(SELECT DISTINCT x '
                 '                       FROM UNNEST(token_array) AS x) AS unique_tokens_per_row '
                 '          FROM SplitTable) '
                 'SELECT token, COUNT(token) as token_count '
                 'FROM TokenTable '
                 'CROSS JOIN UNNEST(TokenTable.unique_tokens_per_row) as token '
                 'WHERE LENGTH(token) > 0 '
                 'GROUP BY token '
                 'ORDER BY token_count DESC, token ASC').format(separator=separator,
                                                                name=col_name,
                                                                table=table_name)
        else:
          # Extract label and frequency
          sql = ('SELECT {name} as token, count(*) as count '
                 'FROM {table} '
                 'WHERE {name} IS NOT NULL '
                 'GROUP BY {name} '
                 'ORDER BY count DESC, token ASC').format(name=col_name,
                                                          table=table_name)
  
        df = _execute_sql(sql, table)
  
        # Save the vocab
        csv_string = df.to_csv(index=False, header=False)
        file_io.write_string_to_file(
            os.path.join(output_dir, constant.VOCAB_ANALYSIS_FILE % col_name),
            csv_string)
        numerical_vocab_stats[col_name] = {'vocab_size': len(df)}
  
        # free memeory
        del csv_string
        del df
      elif transform['transform'] in constant.NUMERIC_TRANSFORMS:
        # get min/max/average
        sql = ('SELECT max({name}) as max_value, min({name}) as min_value, '
               'avg({name}) as avg_value from {table}').format(name=col_name,
                                                               table=table_name)
        df = _execute_sql(sql, table)
        numerical_vocab_stats[col_name] = {'min': df.iloc[0]['min_value'],
                                           'max': df.iloc[0]['max_value'],
                                           'mean': df.iloc[0]['avg_value']}
      sys.stdout.write('column %s analyzed.\n' % col_name)
      sys.stdout.flush()
  
    # get num examples
    sql = 'SELECT count(*) as num_examples from {table}'.format(table=table_name)
    df = _execute_sql(sql, table)
    num_examples = df.iloc[0]['num_examples']
  
    # Write the stats file.
    stats = {'column_stats': numerical_vocab_stats, 'num_examples': num_examples}
    file_io.write_string_to_file(
        os.path.join(output_dir, constant.STATS_FILE),
        json.dumps(stats, indent=2, separators=(',', ': ')))
  
    feature_analysis.save_schema_features(schema, features, output_dir)