create_config_from_excel.py

# -*-encoding:utf-8-*-
# Copyright (c) Alibaba, Inc. and its affiliates.
import logging
import math
import sys

import numpy as np
import pandas as pd

from easy_rec.python.utils import config_util

logging.basicConfig(
    level=logging.INFO, format='[%(asctime)s][%(levelname)s] %(message)s')


class ModelConfigConverter:

  def __init__(self, excel_path, output_path, model_type, column_separator,
               incol_separator, train_input_path, eval_input_path, model_dir):
    self._excel_path = excel_path
    self._output_path = output_path
    self._model_type = model_type
    self._column_separator = column_separator
    self._incol_separator = incol_separator
    self._dict_global = self._parse_global()
    self._tower_dicts = {}
    self._feature_names = []
    self._feature_details = {}
    self._label = ''
    self._train_input_path = train_input_path
    self._eval_input_path = eval_input_path
    self._model_dir = model_dir
    if not self._model_dir:
      self._model_dir = 'experiments/demo'
      logging.warning('model_dir is not specified, set to %s' % self._model_dir)

  def _get_type_name(self, input_name):
    type_dict = {
        'bigint': 'INT64',
        'double': 'DOUBLE',
        'float': 'FLOAT',
        'string': 'STRING',
        'bool': 'BOOL'
    }
    return type_dict[input_name]

  def _get_type_default(self, input_name):
    type_dict = {
        'bigint': '0',
        'double': '0.0',
        'float': '0.0',
        'string': '',
        'bool': 'false'
    }
    return type_dict[input_name]

  def _parse_global(self):
    df = pd.read_excel(self._excel_path, sheet_name='global')
    dict_global = {}
    for i, row in df.iterrows():
      field = {}
      name = field['name'] = row['name'].strip()
      field['type_name'] = row['type']
      field['hash_bucket_size'] = row['hash_bucket_size']
      field['embedding_dim'] = row['embedding_dim']
      field['default_value'] = row['default_value']
      dict_global[name] = field
    return dict_global

  def _add_to_tower(self, tower_name, field):
    if tower_name.lower() == 'nan':
      return
    if tower_name != 'label':
      if self._model_type == 'deepfm':
        if tower_name == 'deep':
          tower_names = ['deep']
        elif tower_name == 'wide':
          tower_names = ['wide']
        elif tower_name == 'wide_and_deep':
          tower_names = ['wide', 'deep']
        else:
          raise ValueError(
              'invalid tower_name[%s] for deepfm model, '
              'only[label, deep, wide, wide_and_deep are supported]' %
              tower_name)
        for tower_name in tower_names:
          if tower_name in self._tower_dicts:
            self._tower_dicts[tower_name].append(field)
          else:
            self._tower_dicts[tower_name] = [field]
      else:
        if tower_name in self._tower_dicts:
          self._tower_dicts[tower_name].append(field)
        else:
          self._tower_dicts[tower_name] = [field]

  def _is_str(self, v):
    if isinstance(v, str):
      return True
    try:
      if isinstance(v, unicode):  # noqa: F821
        return True
    except NameError:
      return False
    return False

  def _parse_features(self):
    df = pd.read_excel(self._excel_path, sheet_name='features')
    for i, row in df.iterrows():
      field = {}
      name = field['name'] = row['name'].strip()
      self._feature_names.append(name)
      field['data_type'] = row['data_type'].strip()
      field['type'] = row['type'].strip()
      g = str(row['global']).strip()

      if g and g != 'nan':
        field['global'] = g

      field['field_name'] = name

      if row['type'].strip() == 'label':
        self._label = name

      if 'global' in field and field['global'] in self._dict_global:
        # 如果是global 有值，就跳过
        def _is_good(v):
          return str(v) not in ['nan', '']

        if _is_good(self._dict_global[field['global']]['default_value']):
          field['default_value'] = self._dict_global[
              field['global']]['default_value']
        if _is_good(self._dict_global[field['global']]['hash_bucket_size']):
          field['hash_bucket_size'] = self._dict_global[
              field['global']]['hash_bucket_size']
        if _is_good(self._dict_global[field['global']]['embedding_dim']):
          field['embedding_dim'] = self._dict_global[
              field['global']]['embedding_dim']
        field['embedding_name'] = field['global']

      for t in [
          'type', 'global', 'hash_bucket_size', 'embedding_dim',
          'default_value', 'weights', 'boundaries'
      ]:
        if t not in row:
          continue
        v = row[t]
        if v not in ['', ' ', 'NaN', np.NaN, np.NAN, 'nan']:
          if self._is_str(v):
            field[t] = v.strip()
          elif not math.isnan(v):
            field[t] = int(v)

        if t == 'default_value' and t not in field:
          field[t] = ''
          if field['type'] == 'dense':
            field[t] = 0.0

      if field['type'] == 'weights':
        field['default_value'] = '1'

      tower_name = row['group']
      if name in self._dict_global:
        field['type'] = 'category'
        field['hash_bucket_size'] = self._dict_global[name]['hash_bucket_size']
        field['embedding_dim'] = self._dict_global[name]['embedding_dim']
        field['default_value'] = self._dict_global[name]['default_value']

      if field['data_type'] == 'bigint':
        field['default_value'] = 0
      elif field['data_type'] == 'double':
        field['default_value'] = 0.0

      if field['type'] not in ['notneed', 'not_need', 'not_needed']:
        tower_name = str(tower_name).strip()
        self._add_to_tower(tower_name, field)
      self._feature_details[name] = field

    # check that tag features weights are one of the fields
    for name, config in self._feature_details.items():
      if config['type'] == 'tags':
        if 'weights' in config and config[
            'weights'] not in self._feature_details:
          raise ValueError(config['weights'] + ' not in field names')

  def _write_train_eval_config(self, fout):
    fout.write('train_input_path: "%s"\n' % self._train_input_path)
    fout.write('eval_input_path: "%s"\n' % self._eval_input_path)
    fout.write("""
    model_dir: "%s"

    train_config {
      log_step_count_steps: 200
      # fine_tune_checkpoint: ""
      optimizer_config: {
        adam_optimizer: {
          learning_rate: {
            exponential_decay_learning_rate {
              initial_learning_rate: 0.0001
              decay_steps: 10000
              decay_factor: 0.5
              min_learning_rate: 0.0000001
            }
          }
        }
      }
      num_steps: 2000
      sync_replicas: true
    }

    eval_config {
      metrics_set: {
           auc {}
      }
    }""" % self._model_dir)

  def _write_deepfm_config(self, fout):
    # write model_config
    fout.write('model_config:{\n')
    fout.write('  model_class: "DeepFM"\n')

    # write feature group configs
    tower_names = list(self._tower_dicts.keys())
    tower_names.sort()
    for tower_name in tower_names:
      fout.write('  feature_groups: {\n')
      fout.write('    group_name: "%s"\n' % tower_name)
      curr_feas = self._tower_dicts[tower_name]
      for fea in curr_feas:
        if fea['type'] == 'weights':
          continue
        fout.write('    feature_names: "%s"\n' % fea['name'])
      fout.write('    wide_deep:%s\n' % tower_name.upper())
      fout.write('  }\n')

    # write deepfm configs
    fout.write("""
      deepfm {
        dnn {
          hidden_units: [128, 64, 32]
        }
        final_dnn {
          hidden_units: [128, 64]
        }
        wide_output_dim: 16
        l2_regularization: 1e-5
      }
      embedding_regularization: 1e-5
    }
    """)

  def _write_multi_tower_config(self, fout):
    # write model_config
    fout.write('model_config:{\n')
    fout.write('  model_class: "MultiTower"\n')

    # write each tower features
    tower_names = list(self._tower_dicts.keys())
    tower_names.sort()
    for tower_name in tower_names:
      fout.write('  feature_groups: {\n')
      fout.write('    group_name: "%s"\n' % tower_name)
      curr_feas = self._tower_dicts[tower_name]
      for fea in curr_feas:
        if fea['type'] == 'weights':
          continue
        fout.write('    feature_names: "%s"\n' % fea['name'])
      fout.write('    wide_deep:DEEP\n')
      fout.write('  }\n')

    # write each tower dnn configs
    fout.write('multi_tower { \n')

    for tower_name in tower_names:
      fout.write("""
      towers {
        input: "%s"
        dnn {
          hidden_units: [256, 192, 128]
        }
      }""" % tower_name)

    fout.write("""
        final_dnn {
          hidden_units: [192, 128, 64]
        }
        l2_regularization: 1e-5
      }
      embedding_regularization: 1e-5
    }""")

  def _write_data_config(self, fout):
    fout.write('data_config {\n')
    fout.write('  separator: "%s"\n' % self._column_separator)
    for name in self._feature_names:
      fout.write('  input_fields: {\n')
      fout.write('    input_name: "%s"\n' % name)
      fout.write('    input_type: %s\n' %
                 self._get_type_name(self._feature_details[name]['data_type']))
      if 'default_value' in self._feature_details[name]:
        fout.write('    default_val:"%s"\n' %
                   self._feature_details[name]['default_value'])
      fout.write('  }\n')

    fout.write('  label_fields: "%s"\n' % self._label)
    fout.write("""
      batch_size: 1024
      prefetch_size: 32
      input_type: CSVInput
    }""")

  def _write_feature_config(self, fout):
    for name in self._feature_names:
      feature = self._feature_details[name]
      if feature['type'] in ['weights', 'notneed', 'label']:
        continue
      if name == self._label:
        continue
      fout.write('feature_configs: {\n')
      fout.write('  input_names: "%s"\n' % name)
      if feature['type'] == 'category':
        fout.write('  feature_type: IdFeature\n')
        fout.write('  embedding_dim: %d\n' % feature['embedding_dim'])
        fout.write('  hash_bucket_size: %d\n' % feature['hash_bucket_size'])
        if 'embedding_name' in feature:
          fout.write('  embedding_name: "%s"\n' % feature['embedding_name'])
      elif feature['type'] == 'dense':
        fout.write('  feature_type: RawFeature\n')
        if self._model_type == 'deepfm':
          assert feature[
              'boundaries'] != '', 'raw features must be discretized by specifying boundaries'
        if 'boundaries' in feature and feature['boundaries'] != '':
          fout.write('  boundaries: [%s]\n' %
                     str(feature['boundaries']).strip())
          fout.write('  embedding_dim: %d\n' % int(feature['embedding_dim']))
      elif feature['type'] == 'tags':
        if 'weights' in feature:
          fout.write('  input_names: "%s"\n' % feature['weights'])
        fout.write('  feature_type: TagFeature\n')
        fout.write('  hash_bucket_size: %d\n' % feature['hash_bucket_size'])
        fout.write('  embedding_dim: %d\n' % feature['embedding_dim'])
        if 'embedding_name' in feature:
          fout.write('  embedding_name: "%s"\n' % feature['embedding_name'])
        fout.write('  separator: "%s"\n' % self._incol_separator)
      elif feature['type'] == 'indexes':
        fout.write('  feature_type: TagFeature\n')
        assert 'hash_bucket_size' in feature
        fout.write('  num_buckets: %d\n' % feature['hash_bucket_size'])
        if 'embedding_dim' in feature:
          fout.write('  embedding_dim: %d\n' % feature['embedding_dim'])
        if 'embedding_name' in feature:
          fout.write('  embedding_name: "%s"\n' % feature['embedding_name'])
        fout.write('  separator: "%s"\n' % self._incol_separator)
      else:
        assert False, 'invalid feature types: %s' % feature['type']
      fout.write('}\n')

  def convert(self):
    self._parse_features()
    logging.info(
        'TOWERS[%d]: %s' %
        (len(self._tower_dicts), ','.join(list(self._tower_dicts.keys()))))
    with open(self._output_path, 'w') as fout:
      self._write_train_eval_config(fout)
      self._write_data_config(fout)
      self._write_feature_config(fout)
      if self._model_type == 'deepfm':
        self._write_deepfm_config(fout)
      elif self._model_type == 'multi_tower':
        self._write_multi_tower_config(fout)
      else:
        logging.warning(
            'the model_config could not be generated automatically, you have to write the model_config manually.'
        )
    # reformat the config
    pipeline_config = config_util.get_configs_from_pipeline_file(
        self._output_path)
    config_util.save_message(pipeline_config, self._output_path)


model_types = ['deepfm', 'multi_tower']

if __name__ == '__main__':
  import argparse

  parser = argparse.ArgumentParser()
  parser.add_argument(
      '--model_type',
      type=str,
      choices=model_types,
      help='model type, currently support: %s' % ','.join(model_types))
  parser.add_argument('--excel_path', type=str, help='excel config path')
  parser.add_argument('--output_path', type=str, help='generated config path')
  parser.add_argument(
      '--column_separator',
      type=str,
      default=',',
      help='column separator, separator betwen features')
  parser.add_argument(
      '--incol_separator',
      type=str,
      default='|',
      help='separator within features, such as tag features')
  parser.add_argument(
      '--train_input_path', type=str, default='', help='train input path')
  parser.add_argument(
      '--eval_input_path', type=str, default='', help='eval input path')
  parser.add_argument('--model_dir', type=str, default='', help='model dir')
  args = parser.parse_args()

  if not args.excel_path or not args.output_path:
    parser.print_usage()
    sys.exit(1)

  logging.info('column_separator = %s in_column_separator = %s' %
               (args.column_separator, args.incol_separator))

  converter = ModelConfigConverter(args.excel_path, args.output_path,
                                   args.model_type, args.column_separator,
                                   args.incol_separator, args.train_input_path,
                                   args.eval_input_path, args.model_dir)
  converter.convert()
  logging.info('Conversion done')
  logging.info('Tips:')
  if args.train_input_path == '' or args.eval_input_path == '':
    logging.info('*.you have to update train_input_path,  eval_input_path')
  logging.info('*.you may need to adjust dnn config or final_dnn config')