update

tasks/crawler_week_report.py

@@ -5,6 +5,8 @@
 # @author : litao
 
 # coding=utf-8
+import json
+
 import pymysql
 import xlwt, datetime
 # from maintenance.func_send_email_with_file import send_file_email
@@ -47,18 +49,13 @@ sparkConf.set("prod.tidb.database", "jerry_prod")
 
 spark = (SparkSession.builder.config(conf=sparkConf).config("spark.sql.extensions", "org.apache.spark.sql.TiExtensions")
 .config("spark.tispark.pd.addresses", "172.16.40.170:2379").appName("LR PYSPARK TEST").enableHiveSupport().getOrCreate())
-ti = pti.TiContext(spark)
 
-ti.tidbMapDatabase("jerry_test")
 spark.sql("ADD JAR hdfs:///user/hive/share/lib/udf/brickhouse-0.7.1-SNAPSHOT.jar")
 spark.sql("ADD JAR hdfs:///user/hive/share/lib/udf/hive-udf-1.0-SNAPSHOT.jar")
 spark.sql("CREATE TEMPORARY FUNCTION json_map AS 'brickhouse.udf.json.JsonMapUDF'")
 spark.sql("CREATE TEMPORARY FUNCTION is_json AS 'com.gmei.hive.common.udf.UDFJsonFormatCheck'")
 spark.sql("CREATE TEMPORARY FUNCTION arrayMerge AS 'com.gmei.hive.common.udf.UDFArryMerge'")
 
-# user_tag3_portrait_df = connectTi(spark,'user_tag3_portrait')
-# user_tag3_portrait_df.createOrReplaceTempView("user_tag3_portrait")
-# user_tag3_portrait_df.show(1,False)
 
 
 es_framework = Elasticsearch(hosts='172.16.32.37', port=9200)
@@ -411,13 +408,15 @@ AND page_name in ('search_home','search_home_more','search_home_welfare','search
     device_df.show(1, False)
     sql_res = device_df.collect()
     for res in sql_res:
-        print(res,type(res))
-        for single_keyword in res.query_words:
+        res_json = json.loads(res.query_words)
+        print(res_json, type(res_json))
+        for single_keyword in res_json:
             data_count = data_dic.get(single_keyword)
             if data_count:
                 data_dic[single_keyword] = data_dic[single_keyword] + 1
             else:
                 data_dic[single_keyword] = 0
+    print(data_dic)
     return data_dic