change test file

monitor.py

@@ -96,40 +96,41 @@ def write_redis(device_id,cid_list):
         print("insert redis fail")
         print(e)
 
-def group_redis(device_id,cid_list):
-    # device_id = x[0]
-    # cid_list = x[1]
-    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 len(tags) != 0:
-        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))
+def group_redis(y):
+    for x in y:
+        device_id = x[0]
+        cid_list = x[1]
+        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()
-        cids = list(set([i[0] for i in result]))
-        if len(cids) >1 :
-            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))
-            else:
-                r.set(key, json.dumps(cids))
-                r.expire(key, 60*60)
-
-
-def group_write(rdd):
-    rdd.groupByKey().foreachPartition(lambda x:x.map())
-    return "good"
+        tags = list(set([i[0] for i in result]))
+        if len(tags) != 0:
+            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()
+            cids = list(set([i[0] for i in result]))
+            if len(cids) >1 :
+                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))
+                else:
+                    r.set(key, json.dumps(cids))
+                    r.expire(key, 60*60)
+
+
+# def group_write(rdd):
+#     rdd.groupByKey().foreachPartition(lambda x:x.map())
+#     return "good"