import pandas as pd import pymysql from datetime import datetime from datetime import timedelta import pickle import time from kafka import KafkaProducer import json from pyspark.streaming.kafka import KafkaUtils from pyspark import SparkContext from pyspark.sql import SQLContext from pyspark.streaming import StreamingContext from pyspark import SparkConf import redis import sys import os import json import pymysql import numpy as np import time import datetime import tensorflow as tf import msgpack import smtplib import requests from email.mime.text import MIMEText from email.utils import formataddr from email.mime.multipart import MIMEMultipart from email.mime.application import MIMEApplication # sys.path.append('/srv/apps/ftrl/Bandist_Streaming') def send_email(app,id,e,extra_information = ''): # 第三方 SMTP 服务 mail_host = 'smtp.exmail.qq.com' # 设置服务器 mail_user = "huangkai@igengmei.com" # 用户名 mail_pass = "UyhVobmDHa4r4ecV" # 口令 sender = 'huangkai@igengmei.com' receivers = ['huangkai@igengmei.com'] # 接收邮件,可设置为你的QQ邮箱或者其他邮箱 e = str(e) msg = MIMEMultipart() part = MIMEText('app_id:'+id+':fail', 'plain', 'utf-8') msg.attach(part) msg['From'] = formataddr(["huangkai", 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.write(extra_information) 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') def ts_cal(): return 0 def cal_ctr(data): a1 = datetime.datetime.now() device_data = data[1] device_id = device_data['device']['device_id'] db_eagle = pymysql.connect(host="172.16.40.158", port=4000, user="root", password="3SYz54LS9#^9sBvC", db="eagle", cursorclass=pymysql.cursors.DictCursor) cursor = db_eagle.cursor() sql = 'select id from online_api_service' cursor.execute(sql) results = cursor.fetchall() device_meigou_ctr_key = 'device_meigou_ctr:device_id:'+str(device_id) device_meigou_params_key = 'device_meigou_params:device_id:'+str(device_id) redis_client = redis.StrictRedis.from_url('redis://:ReDis!GmTx*0aN6@172.16.40.133:6379') meigou_index_dict = dict() meigou_new_params_dict = dict() index_value = 0 init_params_value = 1 model_param_a = list() model_param_b = list() if redis_client.exists(device_meigou_params_key): meigou_params_dict = redis_client.hgetall(device_meigou_params_key) for result in results: if result['id'] in meigou_params_dict.keys(): meigou_index_dict.update({index_value:result['id']}) meigou_new_params_dict.update({result['id']:meigou_index_dict[result['id']]}) model_param_a.append(meigou_params_dict[result['id']]['a']) model_param_b.append(meigou_params_dict[result['id']]['b']) index_value += 1 else: meigou_index_dict.update({index_value: result['id']}) meigou_new_params_dict.update({result['id']:{"a":init_params_value,"b":init_params_value}}) model_param_a.append(init_params_value) model_param_b.append(init_params_value) index_value +=1 else: for result in results: meigou_new_params_dict.update({result['id']:{"a":init_params_value,"b":init_params_value}}) meigou_index_dict.update({index_value: result['id']}) model_param_a.append(init_params_value) model_param_b.append(init_params_value) index_value += 1 a2 = datetime.datetime.now() num_actions = len(results) user_feature = np.array([1]) # hparams_nlinear = tf.contrib.training.HParams(num_actions=num_actions, # context_dim=1, # init_scale=0.3, # activation=tf.nn.relu, # layer_sizes=[1], # batch_size=1, # activate_decay=True, # initial_lr=0.1, # max_grad_norm=5.0, # show_training=False, # freq_summary=1000, # buffer_s=-1, # initial_pulls=0, # reset_lr=True, # lr_decay_rate=0.5, # training_freq=1, # training_freq_network=10000, # training_epochs=100, # a0=model_param_a, # b0=model_param_b, # lambda_prior=0.25) # inital model model = NeuralLinearPosteriorSampling('NeuralLinear',num_actions,model_param_a,model_param_b) a2 =datetime.datetime.now() vals = model.action(user_feature) # model.update(user_feature,0,np.array(1)) max =vals.max() min = vals.min() ctr_0_1 = (vals-min)/(max-min) meigou_ctr_dict = dict() a3 =datetime.datetime.now() for i in range(len(ctr_0_1)): meigou_ctr_dict.update({meigou_index_dict[i]:ctr_0_1[i]}) redis_client.set(device_meigou_ctr_key,json.dumps(meigou_ctr_dict)) a4 = datetime.datetime.now() send_email(str(a1),str(a2),str(a3),str(a4)) def choose_action(): return 0 def Filter_Data(data): data_dict = data[1] if b'content' in data_dict: return False elif 'type' in data_dict: if data_dict['type'] == 'device_opened' and data_dict['device']['device_id'] == '8E699605-DC2A-46B6-8B47-E9E809353055': return True def write_to_kafka(): producer = KafkaProducer(bootstrap_servers=["172.16.44.25:9092","172.16.44.31:9092","172.16.44.45:9092"], key_serializer=lambda k: json.dumps(k).encode('utf-8'), value_serializer=lambda v: json.dumps(v).encode('utf-8')) future = producer.send(topic="test_topic", key="hello", value="world") try: record_metadata = future.get(timeout=10) print("send ok") except kafka_errors as e: print(str(e)) def Ctr(rdd): try: results = rdd write_to_kafka() return results except: print("fail") def m_decoder(s): if s is None: return None try: data = json.loads(s) return data except: data = msgpack.loads(s, encoding='utf-8') return data if __name__ == '__main__': # Spark-Streaming-Kafka sc = SparkContext(conf=SparkConf().setMaster("spark://nvwa01:7077").setAppName("kafka_test") .set("spark.io.compression.codec", "lzf")) ssc = SQLContext(sc) ssc = StreamingContext(sc, 10) sc.setLogLevel("WARN") kafkaParams = {"metadata.broker.list": "172.16.44.25:9092,172.16.44.31:9092,172.16.44.45:9092", "group.id": "kafka_test", "socket.timeout.ms": "600000", "auto.offset.reset": "largest"} stream = KafkaUtils.createDirectStream(ssc, ["test_topic"], kafkaParams, keyDecoder=m_decoder, valueDecoder=m_decoder) transformstream = stream.transform(lambda x: Ctr(x)) transformstream.pprint() ssc.start() ssc.awaitTermination()