change test file

monitor.py

 from __future__ import absolute_import
 from __future__ import division
 from __future__ import print_function
+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 msgpack
 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 numpy as np
+import pandas as pd
+import time
 import datetime
 
-
-# filter logging
-def gbk_decoder(s):
-    if s is None:
-        return None
-    try:
-        data = msgpack.loads(s,encoding='utf-8')
-        return data
-    except:
-        data = json.loads(s)
-        return data
-
-
-def maidian(x):
-    try:
-        data = x[1]
+def Json(x):
+    data = json.loads(x[1])
     if 'type' in data and 'device' in data and 'params' in data and 'card_content_type' in data['params']:
-            if data['type'] == 'on_click_card' and data['params']['card_content_type'] == 'diary':
-                # and data["device"]['device_id'] == "E417C286-40A4-42F6-BDA9-AEEBD8FEC3B6"
+        if data['type'] == 'on_click_card'and data['params']['card_content_type'] =='diary':
+            return True
+        # elif data['type'] == 'on_click_button'and 'params' in data:
+        #     if 'page_name' in data['params'] and 'button_name' in data['params'] and 'extra_params' in data['params']:
+        #         if data['params']['page_name'] =='page_choose_interest' and data['params']['button_name']=='next':
+        #             return True
+
             return True
         else:
             return False
     else:
         return False
-
-    except Exception as e:
-        print("filter fail")
-        print(e)
-
-
-def get_data(x):
-    try:
-        device_id = x[1]['device']['device_id']
-        diary_id = x[1]['params']["card_id"]
-        return device_id,diary_id
-    except Exception as e:
-        print("get_data fail")
-        send_email("get_data", "get_data", e)
-
-
-def write_redis(device_id,cid_list):
-    try:
-        db = pymysql.connect(host='172.16.40.158', port=4000, user='root', passwd='3SYz54LS9#^9sBvC', db='eagle')
-        sql = "select b.id from src_mimas_prod_api_diary_tags a left join src_zhengxing_api_tag b " \
-              "on a.tag_id = b.id where b.tag_type = '3' and a.diary_id in {}".format(tuple(cid_list))
-        cursor = db.cursor()
-        cursor.execute(sql)
-        result = cursor.fetchall()
-        tags = list(set([i[0] for i in result]))
-        if tags is not None:
-            sql = "select a.id from src_mimas_prod_api_diary a left join src_mimas_prod_api_diary_tags b " \
-                  "on a.id=b.diary_id left join src_zhengxing_api_tag c on b.tag_id=c.id " \
-                  "where a.is_online = 1 and a.content_level >= '3' " \
-                  "and c.id in {} and c.tag_type = '3'".format(tuple(tags))
-            cursor.execute(sql)
-            result = cursor.fetchall()
-            if result is not None:
-                cids = list(set([i[0] for i in result]))
-                r = redis.StrictRedis.from_url('redis://:ReDis!GmTx*0aN6@172.16.40.133:6379')
-                key = str(device_id) + "_dislike_diary"
-                if r.exists(key):
-                    value = eval(r.get(key))
-                    value.extend(cids)
-                    cids = json.dumps(list(set(value)))
-                    r.set(key, json.dumps(cids))
-                    print("insert one")
-                else:
-                    r.set(key, json.dumps(cids))
-                    r.expire(key, 60*60)
-                    print("insert two")
-    except Exception as e:
-        print("insert redis fail")
-        print(e)
-
-
+#queue
+#rdd transform
 def model(rdd):
     try:
-        rdd.filter(lambda x: maidian(x))
-            # .map(lambda x:get_data(x))
-            #                                  .na.drop().groupByKey())\
-            # .map(lambda x:write_redis(x[0],x[1]))
-    except Exception as e:
+        rdd = rdd.filter(lambda x:Json(x))
+            # .repartition(10)
+        return  rdd
+    except:
         print("fail")
-        print(e)
-
 
-if __name__ == '__main__':
-    sc = SparkContext(conf=SparkConf().setMaster("spark://nvwa01:7077").setAppName("dislike_filter").set(
-        "spark.io.compression.codec", "lzf"))
-    ssc = StreamingContext(sc, 6)
-    sc.setLogLevel("WARN")
-    kafkaParams = {"metadata.broker.list": "172.16.44.25:9092,172.16.44.31:9092,172.16.44.45:9092",
-                   "group.id": "dislike",
+# Spark-Streaming-Kafka
+sc = SparkContext(conf=SparkConf().setMaster("spark://nvwa01:7077").setAppName("filter").set("spark.io.compression.codec", "lzf"))
+ssc=SQLContext(sc)
+ssc = StreamingContext(sc,4)
+sc.setLogLevel("WARN")
+kafkaParams = {"metadata.broker.list": "172.16.44.25:9092,172.16.44.31:9092,172.16.44.45:9092",
+               "group.id": "filter",
                "socket.timeout.ms": "600000",
                "auto.offset.reset": "largest"}
-    try:
-        stream = KafkaUtils.createDirectStream(ssc, ["gm-maidian-data"], kafkaParams, keyDecoder=gbk_decoder,
-                                               valueDecoder=gbk_decoder)
-        transformstream = stream.transform(lambda x: model(x))
-        transformstream.pprint()
-        ssc.start()
-        ssc.awaitTermination()
-    except Exception as e:
-        print(e)
-        # send_email(sc.appName, sc.applicationId, e)
 
 
+stream = KafkaUtils.createDirectStream(ssc, ["gm-maidian-data"], kafkaParams)
+transformstream = stream.transform(lambda x:model(x))
+transformstream.pprint()
+ssc.start()
+ssc.awaitTermination()
\ No newline at end of file