text: #  Licensed to the Apache Software Foundation (ASF) under one
  #  or more contributor license agreements.  See the NOTICE file
  #  distributed with this work for additional information
  #  regarding copyright ownership.  The ASF licenses this file
  #  to you under the Apache License, Version 2.0 (the
  #  "License"); you may not use this file except in compliance
  #  with the License.  You may obtain a copy of the License at
  
  #  http://www.apache.org/licenses/LICENSE-2.0.html
  
  #  Unless required by applicable law or agreed to in writing, software
  #  distributed under the License is distributed on an "AS IS" BASIS,
  #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  #  See the License for the specific language governing permissions and
  #  limitations under the License.
  
  from pyspark import SparkContext
  from pyspark.sql import HiveContext
  import yaml
  import argparse
  
  '''
  input: logfile
  output: lookalike_seeduser table
  
  
  '''
  
  
  def run(hive_context, cfg, kwi):
      seed_user_table = cfg['input']['seeduser_table']
      log_table = cfg['input']['log_table']
      number_of_seeduser = cfg['input']['number_of_seeduser']
  
      # command = "select * from (select * from {} where is_click=1 and keyword_index=29) as s join (select * from {} where is_click=1 and keyword_index=26) as b on b.did = s.did where s.gender = 1"
      command = "SELECT * FROM {} WHERE is_click=1 AND keyword_index={}"
      df = hive_context.sql(command.format(log_table, kwi))
      user_list = df.select('did').alias('did').distinct().limit(number_of_seeduser)
      user_list.cache()
  
      user_list.write.option("header", "true").option(
          "encoding", "UTF-8").mode("overwrite").format('hive').saveAsTable(seed_user_table)
  
  
  if __name__ == "__main__":
      """
      select seed users
      """
      parser = argparse.ArgumentParser(description=" ")
      parser.add_argument('config_file')
      parser.add_argument('kwi')
      args = parser.parse_args()
      kwi = args.kwi
      with open(args.config_file, 'r') as yml_file:
          cfg = yaml.safe_load(yml_file)
  
      sc = SparkContext.getOrCreate()
      sc.setLogLevel('WARN')
      hive_context = HiveContext(sc)
  
      run(hive_context=hive_context, cfg=cfg, kwi=kwi)
      sc.stop()
  