change

ctr.py

@@ -179,6 +179,7 @@ def write_to_kafka():
 def Ctr(rdd):
     try:
         results = rdd
+        write_to_kafka()
         return results
     except:
         print("fail")
@@ -195,7 +196,7 @@ def m_decoder(s):
 
 if __name__ == '__main__':
     # Spark-Streaming-Kafka
-    write_to_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)