统计like项目

eda/feededa/src/main/scala/com/gmei/temp_analysis.scala

@@ -604,10 +604,9 @@ object alpha_ctr {
            |and partition_day='${partition_date}'
        """.stripMargin
       )
-      val a =expoure_cards.rdd.map(row => row(0).toString).map(row=>{
-        val lenth=parse_json(row)
-        (stat_date,lenth)
-      }).reduceByKey((x,y) => x + y).toDF("stat_date","exposure_count_recommend")
+      val a =expoure_cards.rdd.map(row => row(0).toString).map(row=>parse_json(row)).collect().sum
+      val result1=List((stat_date,a))
+      val df1 = sc.createDataFrame(result1).toDF("stat_date","expoure_count_recommend")
 
       val expoure_cards2=sc.sql(
         s"""
@@ -619,15 +618,15 @@ object alpha_ctr {
            |and partition_day='${partition_date}'
        """.stripMargin
       )
-      val b =expoure_cards2.rdd.map(row => row(0).toString).map(row=>{
-        val lenth=parse_json(row)
-        (stat_date,lenth)
-      }).reduceByKey((x,y) => x + y).toDF("stat_date","exposure_count_focus")
+      val b =expoure_cards2.rdd.map(row => row(0).toString).map(row=>parse_json(row)).collect().sum
+      val result2=List((stat_date,b))
+      val df2 = sc.createDataFrame(result2).toDF("stat_date","expoure_count_focus")
+
 
 
       val result=click_count_recommend.join(click_count_focus,"stat_date")
-        .join(a,"stat_date")
-        .join(b,"stat_date")
+        .join(df1,"stat_date")
+        .join(df2,"stat_date")