colloct to master and pritn

src/main/scala/com/gmei/data/dq/pvCheker.scala

@@ -8,7 +8,7 @@ case class Record(page_name: String, flag: Int, cl_type: String, count: Int)
 
 object pvReferCheker {
 
-  def validateRefer(page_name: String, d: Iterable[(Int, String, Int)]) = {
+  def validateRefer(page_name: String, d: Iterable[(Int, String, Long)]): String = {
     // demo data
     /*
       about_me_message_list	0	ios	335
@@ -25,19 +25,21 @@ object pvReferCheker {
       all_sort	0	android	639
      */
     if (d.size == 1 || (d.size == 2 && d.head._1 == d.last._1)) {
-      println(s"$page_name seems good!")
+      s"$page_name seems good!"
     } else if (d.size == 3) {
       val ps = d.partition(_._2 == "ios")
       if (ps._1.size == 1)
-        println(s"$page_name ${ps._2.head._2} client seems bad!")
+        s"$page_name ${ps._2.head._2} client seems bad!"
       else
-        println(s"$page_name ${ps._1.head._2} client seems bad!")
+        s"$page_name ${ps._1.head._2} client seems bad!"
     } else if (d.size == 4) {
       val ps = d.partition(_._2 == "ios")
       if (ps._1.head._3 / ps._1.last._3 > ps._2.head._3 / ps._1.last._3)
-        println(s"$page_name client ${ps._1.head._2} seems bad!")
+        s"$page_name client ${ps._1.head._2} seems bad!"
       else
-        println(s"$page_name client ${ps._2.head._2} seems bad!")
+        s"$page_name client ${ps._2.head._2} seems bad!"
+    } else {
+      s"$page_name ok"
     }
   }
 
@@ -47,31 +49,31 @@ object pvReferCheker {
     val vault: Int = 0
     val x = sc.sql(
       s"""
-        |select * from (
-        | select
-        |   params['page_name'] as page_name,
-        |   (case when params['referrer'] = '' or params['referrer'] is null then 0 else 1 end) as has_referrer,
-        |   cl_type,
-        |   count(1) as c,
-        | from
-        |   online.tl_hdfs_maidian_view
-        | where
-        |   partition_date = '$partition_date' and action = 'page_view'
-        | group by
-        |   params['page_name'],
-        |   (case when params['referrer'] = '' or params['referrer'] is null then 0 else 1 end),
-        |   cl_type
-        |) c
-        |order by c.page_name
+        |select
+        |  params['page_name'] as page_name,
+        |  (case when params['referrer'] = '' or params['referrer'] is null then 0 else 1 end) as has_referrer,
+        |  cl_type,
+        |  count(1) as c
+        |from
+        |  online.bl_hdfs_maidian_updates
+        |where
+        |  partition_date = '$partition_date' and action = 'page_view'
+        |group by
+        |  params['page_name'],
+        |  (case when params['referrer'] = '' or params['referrer'] is null then 0 else 1 end),
+        |  cl_type
+        |order by params['page_name']
       """.stripMargin)
 
     val y = x.rdd.map {
-      case Row(page_name: String, has_referrer: Int, cl_type: String, c: Int) => {
+      case Row(page_name: String, has_referrer: Int, cl_type: String, c: Long) => {
         page_name -> (has_referrer, cl_type, c)
       }
     }
+
     val z = y.groupByKey().map {
       case (p, v) => validateRefer(p, v)
     }
+    z.collect.foreach {println}
   }
 }