change test file

monitor.py

@@ -96,16 +96,14 @@ def write_redis(device_id,cid_list):
         print("insert redis fail")
         print(e)
 
-def group_redis(y):
-    for x in y:
-        device_id = x[0]
-        cid_list = x[1]
+def group_redis(device_id,cid_list):
     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()
+    db.close()
     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 " \
@@ -147,7 +145,7 @@ kafkaParams = {"metadata.broker.list": "172.16.44.25:9092,172.16.44.31:9092,172.
 
 stream = KafkaUtils.createDirectStream(ssc, ["gm-maidian-data"], kafkaParams)
 transformstream = stream.transform(lambda x:model(x)).foreachRDD(lambda x:x.groupByKey().
-                                                                 foreachPartition(group_redis))
+                                                                 foreach(group_redis))
 # transformstream.pprint()
 # print(transformstream)
 ssc.start()