change test

monitor.py

@@ -30,8 +30,8 @@ def Json(x):
 
 def model(rdd):
     try:
-        rdd = rdd.filter(lambda x:Json(x)).repartition(10).map(lambda x:get_data(x))
-            # .map(lambda x:write_redis(x[0],x[1],x[2]))
+        rdd = rdd.filter(lambda x:Json(x)).repartition(10).map(lambda x:get_data(x))\
+            .map(lambda x:write_redis(x[0],x[1],x[2]))
         return rdd
     except:
         print("fail")
@@ -77,7 +77,7 @@ def tractate_write(device_id, cid):
                 db.close()
 
                 if len(result) > 0:
-                    cids = [i[0] for i in result]
+                    cids = [str(i[0]) for i in result]
 
                     r = redis.Redis(host="172.16.40.135", port=5379, password="",db = 2)
                     key = str(device_id) + "_dislike_tractate"
@@ -117,7 +117,7 @@ def question_write(device_id,cid):
                 result = cursor.fetchall()
                 db.close()
                 if len(result) > 0:
-                    cids = [i[0] for i in result]
+                    cids = [str(i[0]) for i in result]
 
                     r = redis.Redis(host="172.16.40.135", port=5379, password="", db=2)
                     key = str(device_id) + "_dislike_qa"
@@ -158,7 +158,7 @@ def diary_write(device_id,cid):
                 result = cursor.fetchall()
                 db.close()
                 if len(result) > 0:
-                    cids = [i[0] for i in result]
+                    cids = [str(i[0]) for i in result]
                     r = redis.Redis(host="172.16.40.135", port=5379, password="", db=2)
                     key = str(device_id) + "_dislike_diary"
                     if r.exists(key):
@@ -176,23 +176,21 @@ def diary_write(device_id,cid):
         print(e)
 
 
-# sc = SparkContext(conf=SparkConf().setMaster("spark://nvwa01:7077").setAppName("dislike").set("spark.io.compression.codec", "lzf"))
-# 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": "dislike",
-#                "socket.timeout.ms": "600000",
-#                "auto.offset.reset": "largest"}
-#
-#
-# stream = KafkaUtils.createDirectStream(ssc, ["gm-maidian-data"], kafkaParams)
-# transformstream = stream.transform(lambda x:model(x))
-# transformstream.pprint()
-#
-# ssc.start()
-# ssc.awaitTermination()
-
-question_write("hello",5)
+sc = SparkContext(conf=SparkConf().setMaster("spark://nvwa01:7077").setAppName("dislike").set("spark.io.compression.codec", "lzf"))
+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": "dislike",
+               "socket.timeout.ms": "600000",
+               "auto.offset.reset": "largest"}
+
+
+stream = KafkaUtils.createDirectStream(ssc, ["gm-maidian-data"], kafkaParams)
+transformstream = stream.transform(lambda x:model(x))
+transformstream.pprint()
+
+ssc.start()
+ssc.awaitTermination()