update

task/search_strategy_d.py

@@ -20,6 +20,7 @@ import datetime
 import time
 from pyspark import SparkConf
 from pyspark.sql import SparkSession, DataFrame
+
 # from pyspark.sql.functions import lit
 # import pytispark.pytispark as pti
 
@@ -80,7 +81,7 @@ for t in range(1, task_days):
     today_str = now.strftime("%Y%m%d")
     yesterday_str = (now + datetime.timedelta(days=-1)).strftime("%Y%m%d")
     one_week_age_str = (now + datetime.timedelta(days=-7)).strftime("%Y%m%d")
-    sql_distinct_device_id = """
+    sql_spam_pv_device_id = """
 
     SELECT DISTINCT device_id
     FROM ml.ml_d_ct_dv_devicespam_d  --去除机构刷单设备，即作弊设备（浏览和曝光事件去除）
@@ -90,27 +91,36 @@ for t in range(1, task_days):
     SELECT DISTINCT dev.device_id 
     FROM dim.dim_device_user_staff   --去除内网用户
 
-    UNION ALL 
-    (
-        SELECT t1.device_id
+   	) """.format(yesterday_str=yesterday_str, today_str=today_str)
+    print(sql_spam_pv_device_id)
+    spam_pv_df = spark.sql(sql_spam_pv_device_id)
+    spam_pv_view = spam_pv_df.createOrReplaceTempView("spam_pv")
+    spam_pv_df.show(1)
+    sql_res = spam_pv_df.collect()
+    for res in sql_res:
+        print(res)
+
+    print("-------------------------------")
+    sql_dev_device_id = """
+        SELECT partition_date,device_id
 			FROM 
 			(--找出user_id当天活跃的第一个设备id
 			    SELECT user_id,partition_date,
 			            if(size(device_list) > 0, device_list [ 0 ], '') AS device_id
 			      FROM online.ml_user_updates
-              WHERE partition_date>='${{yesterday_str}}' AND partition_date<'${{today_str}}'
+			      WHERE partition_date>='{yesterday_str}' AND partition_date<'{today_str}'
 			)t1
 			JOIN
 			(  --医生账号
 			    SELECT distinct user_id
 			    FROM online.tl_hdfs_doctor_view
-            WHERE partition_date = {yesterday_str}
+			    WHERE partition_date = '{yesterday_str}'
 
 			    --马甲账号/模特用户
 			    UNION ALL
 			    SELECT user_id
 			    FROM ml.ml_c_ct_ui_user_dimen_d
-            WHERE partition_day = {yesterday_str}
+			    WHERE partition_day = '{yesterday_str}'
 			    AND (is_puppet = 'true' or is_classifyuser = 'true')
 
 			    UNION ALL
@@ -126,25 +136,25 @@ for t in range(1, task_days):
 			        SELECT user_id, v.device_id as device_id
 			        FROM online.ml_user_history_detail
 			            LATERAL VIEW EXPLODE(device_history_list) v AS device_id
-                WHERE partition_date = {yesterday_str}
+			        WHERE partition_date = '{yesterday_str}'
 			    )t1
 			    JOIN
 			    (
 			        SELECT device_id
 			        FROM online.ml_device_history_detail
-                WHERE partition_date = {yesterday_str}
+			        WHERE partition_date = '{yesterday_str}'
 			        AND is_login_doctor = '1'
 			    )t2
 			        ON t1.device_id = t2.device_id
 			)t2
 			on t1.user_id=t2.user_id
-        group by device_id
-		)dev """.format(yesterday_str=yesterday_str,today_str=today_str)
-    print(sql_distinct_device_id)
-    distinct_device_id_df = spark.sql(sql_distinct_device_id,)
-    distinct_device_id_view = distinct_device_id_df.createOrReplaceTempView("distinct_device_id_view")
-    distinct_device_id_df.show(1)
-    sql_res = distinct_device_id_df.collect()
+			group by partition_date,device_id
+       	) """.format(yesterday_str=yesterday_str, today_str=today_str)
+    print(sql_dev_device_id)
+    dev_df = spark.sql(sql_dev_device_id)
+    dev_df_view = spam_pv_df.createOrReplaceTempView("dev_view")
+    dev_df_view.show(1)
+    sql_res = dev_df_view.collect()
     for res in sql_res:
         print(res)
 
@@ -257,10 +267,12 @@ FROM
 		)t2
 		on t1.cl_id=t2.device_id AND t1.partition_date = t2.partition_date
 
-		LEFT JOIN distinct_device_id_view
-		on  t1.partition_date=distinct_device_id_view.partition_date AND t1.cl_id=distinct_device_id_view.device_id
-		WHERE (distinct_device_id_view.dev_device_id IS NULL or distinct_device_id_view.dev_device_id ='')
-		GROUP BY t1.partition_date,t2.active_type,device_os_type,channel
+		LEFT JOIN spam_pv on spam_pv.device_id=t1.cl_id
+		LEFT JOIN dev_view
+		on t1.partition_date=dev_view.partition_date and t1.cl_id=dev_view.device_id
+		WHERE (spam_pv.device_id IS NULL or spam_pv.device_id ='')
+		and (dev_view.device_id is null or dev_view.device_id ='')
+		GROUP BY t1.partition_date,active_type,device_os_type,channel
 	)t
 )t3
 
@@ -349,9 +361,11 @@ LEFT JOIN
         LATERAL VIEW explode(mas.active_type) t2 AS active_type
 	)dev
 	    on t1.cl_id=dev.device_id and t1.partition_date = dev.partition_date
-	LEFT JOIN distinct_device_id_view
-	on t1.cl_id=distinct_device_id_view.device_id
-	WHERE (distinct_device_id_view.dev_device_id IS NULL or distinct_device_id_view.dev_device_id ='')
+		LEFT JOIN spam_pv on spam_pv.device_id=t1.cl_id
+		LEFT JOIN dev_view
+		on t1.partition_date=dev_view.partition_date and t1.cl_id=dev_view.device_id
+		WHERE (spam_pv.device_id IS NULL or spam_pv.device_id ='')
+		and (dev_view.device_id is null or dev_view.device_id ='')
 		GROUP BY t1.partition_date,active_type,device_os_type,channel
 )t4
 on t3.partition_date=t4.partition_date and t3.active_type=t4.active_type and t3.device_os_type = t4.device_os_type AND t3.channel = t4.channel