add dist update user portrait

eda/smart_rank/dist_update_user_portrait.py

+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+import pymysql
+import smtplib
+from email.mime.text import MIMEText
+from email.utils import formataddr
+from email.mime.multipart import MIMEMultipart
+from email.mime.application import MIMEApplication
+import redis
+import datetime
+from pyspark import SparkConf
+import time
+from pyspark.sql import SparkSession
+import json
+import numpy as np
+import pandas as pd
+
+
+def compute_henqiang(x):
+    score = 15-x*((15-0.5)/180)
+    if score>0.5:
+        return score
+    else:
+        return 0.5
+def compute_jiaoqiang(x):
+    score = 12-x*(12/180)
+    if score>0.5:
+        return score
+    else:
+        return 0.5
+def compute_ruoyixiang(x):
+    score = 5-x*((5-0.5)/180)
+    if score>0.5:
+        return score
+    else:
+        return 0.5
+def compute_validate(x):
+    score = 10-x*((10-0.5)/180)
+    if score>0.5:
+        return score
+    else:
+        return 0.5
+def tag_list2dict(lst,size):
+    result = []
+    if lst:
+        for i in lst:
+            tmp = dict()
+            tmp["content"] = i["tag_id"]
+            if isinstance(i['tag_id'], int):
+                tmp["type"] = "tag"
+            else:
+                tmp["type"] = "search_word"
+            tmp["score"] = i["tag_score"]
+            result.append(tmp)
+    return result[:size]
+
+
+def get_user_tag_score(cl_id, all_log_df):
+    user_log_df = all_log_df.loc(all_log_df['cl_id'] == cl_id)
+    if not user_log_df.empty:
+        user_log_df["tag_id"] = np.where(user_log_df["action"] == "do_search",user_log_df["tag_referrer"],user_log_df["tag_id"])
+        user_log_df["days_diff_now"] = round((int(time.time())-user_log_df["time"]) / (24*60*60))
+        user_log_df["tag_score"] = user_log_df.apply(
+            lambda x: compute_henqiang(x.days_diff_now) if x.score_type == "henqiang" else (
+                compute_jiaoqiang(x.days_diff_now) if x.score_type == "jiaoqiang" else (
+                    compute_ruoyixiang(x.days_diff_now) if x.score_type == "ruoyixiang" else compute_validate(x.days_diff_now))), axis=1)
+        finally_score = user_log_df.sort_values(by=["tag_score","time"],ascending=False)
+        finally_score.drop_duplicates(subset="tag_id", inplace=True)
+        finally_score_lst = finally_score[["tag_id","tag_score"]].to_dict('record')
+        tag_id_list = tag_list2dict(finally_score_lst,size)
+        return cl_id, tag_id_list
+    else:
+        return ()
+
+
+def send_email(app,id,e):
+    # 第三方 SMTP 服务
+    mail_host = 'smtp.exmail.qq.com'  # 设置服务器
+    mail_user = "gaoyazhe@igengmei.com"  # 用户名
+    mail_pass = "VCrKTui99a7ALhiK"  # 口令
+
+    sender = 'gaoyazhe@igengmei.com'
+    receivers = ['gaoyazhe@igengmei.com']  # 接收邮件，可设置为你的QQ邮箱或者其他邮箱
+    e = str(e)
+    msg = MIMEMultipart()
+    part = MIMEText('app_id:'+id+':fail', 'plain', 'utf-8')
+    msg.attach(part)
+    msg['From'] = formataddr(["gaoyazhe", sender])
+    # 括号里的对应收件人邮箱昵称、收件人邮箱账号
+    msg['To'] = ";".join(receivers)
+    # message['Cc'] = ";".join(cc_reciver)
+
+    msg['Subject'] = 'spark streaming:app_name:'+app
+    with open('error.txt','w') as f:
+         f.write(e)
+         f.close()
+    part = MIMEApplication(open('error.txt', 'r').read())
+    part.add_header('Content-Disposition', 'attachment', filename="error.txt")
+    msg.attach(part)
+
+    try:
+        smtpObj = smtplib.SMTP_SSL(mail_host, 465)
+        smtpObj.login(mail_user, mail_pass)
+        smtpObj.sendmail(sender, receivers, msg.as_string())
+    except smtplib.SMTPException:
+        print('error')
+
+
+
+
+if __name__ == '__main__':
+    try:
+        db_jerry_test = pymysql.connect(host='172.16.40.158', port=4000, user='root', passwd='3SYz54LS9#^9sBvC',
+                                        db='jerry_test', charset='utf8')
+        cur_jerry_test = db_jerry_test.cursor()
+
+        # 获取所有用户的设备id
+        sql_device_ids = "select distinct cl_id from user_new_tag_log"
+        cur_jerry_test.execute(sql_device_ids)
+        device_ids_lst = [i[0] for i in cur_jerry_test.fetchall()]
+
+        # 获取所有用户的行为日志
+        sql_all_log = "select time,cl_id,score_type,tag_id,tag_referrer,action from user_new_tag_log"
+        cur_jerry_test.execute(sql_all_log)
+        all_log = cur_jerry_test.fetchall()
+        db_jerry_test.close()
+        all_log_df = pd.DataFrame(list(all_log))
+        all_log_df.columns = ["time", "cl_id", "score_type","tag_id","tag_referrer","action"]
+        # rdd
+        sparkConf = SparkConf().set("spark.hive.mapred.supports.subdirectories", "true") \
+            .set("spark.hadoop.mapreduce.input.fileinputformat.input.dir.recursive", "true") \
+            .set("spark.tispark.plan.allow_index_double_read", "false") \
+            .set("spark.tispark.plan.allow_index_read", "true") \
+            .set("spark.sql.extensions", "org.apache.spark.sql.TiExtensions") \
+            .set("spark.tispark.pd.addresses", "172.16.40.158:2379").set("spark.io.compression.codec", "lzf") \
+            .set("spark.driver.maxResultSize", "8g").set("spark.sql.avro.compression.codec", "snappy")
+
+        spark = SparkSession.builder.config(conf=sparkConf).enableHiveSupport().getOrCreate()
+        spark.sparkContext.setLogLevel("WARN")
+
+        device_ids_lst_rdd = spark.sparkContext.parallelize(device_ids_lst)
+        gm_kv_cli = redis.Redis(host="172.16.40.135", port=5379, db=6, socket_timeout=2000)
+        result = device_ids_lst_rdd.repartition(40).map(lambda x: get_user_tag_score(x, all_log_df))
+        result.take(10).foreach(print)
+    except Exception as e:
+        print(e)
\ No newline at end of file