Apache Spark 批处理作业优化

Question

我是 Scala 和 Spark 的新手，在编写从 MR 作业到 Spark 的迁移代码时遇到了一些问题。基本上我需要帮助来重构/优化我到目前为止编写的代码，以及确定进一步完成任务的方法。

我的问题如下：

1. RDD 空检查的语句if (!usefulData.isEmpty()) 能否以其他方式实现，例如在过滤器函数中。只有在过滤器的输出返回一些行时才应该调用映射或归约/聚合函数（基本上以某种方式使用 scala Option 类）

2. 如果您看到代码，要生成每个 KPI，我必须对 usefulData 进行一次又一次的映射操作，这是过滤和解析的文件数据 RDD。这是一个好方法吗，我们可以将一些地图操作合二为一吗？

3. 我应该将输入文件中的 Spitted 字段放入具有 16 个索引的元组中，还是有其他更好/有效的方法？

4. 如果你看到KPI，success_cnt, success_uu, failure_cnt and failure_uu，它们的处理几乎相同，我想为它们创建一个通用的方法。但无法这样做，我无法将元组索引作为输入传递给函数内部的映射

5. 最后但最重要的是，我想要一种最终方法来组合所有这些 KPI，就像我们在 MR 中的 HashMap 中所做的那样，并将其写入那里的上下文（基本上面临问题，因为在 MR 中，map 和 reduce 函数从 InputSplit 处理单行，但这里使用转换一次处理整个文件，因此我无法创建类级别变量并写入 POJO 类，然后在 Context 中聚合）

   我的代码如下：

---

    import org.apache.log4j.{Level, Logger}
    import org.apache.spark.storage.StorageLevel
    import org.apache.spark.{SparkConf, SparkContext}
    import org.apache.spark.sql.SQLContext
    import org.apache.spark.rdd._
    
    /**
      * Sample Data : 1 Line
      * 11685147    com.my.record.xxxxxx    2015-12-27      200 RECOVER <MacBookAir4,2> <Mac OS X;10.11.2;22222> <com.my.site>      0   com.my.os   1451248910234   1   10
      *
      * Parsed Data :
      * 
      * l._1 : 1685147
      * l._2 : com.my.record.xxxxxx
      * l._3 : 2015-12-27
      * l._4 :
      * l._5 : 200
      * l._6 : RECOVER
      * l._7 : MacBookAir4
      * l._8 : 22222
      * l._9 : Mac OS X
      * l._10 : 10.11.2
      * l._11 :
      * l._12 : 0/-6015
      * l._13 : com.my.os
      * l._14 : 1451248910234
      * l._15 : 1
      * l._16 : 10
      *
      **/
    
    object MySparkApp {
    
      def main(args: Array[String]) {
    
        Logger.getLogger("org").setLevel(Level.OFF)
        Logger.getLogger("akka").setLevel(Level.OFF)
    
        val conf = new SparkConf().setMaster("local").setAppName("My-Spark-App").set("spark.executor.memory", "2G").set("spark.driver.memory", "1G")
        val sc = new SparkContext(conf)
    
        val pattern = "[a-zA-Z]+".r
    
        def parseClientInfo(field: String) = {
          val fields = field.split(">")
    
          val devType = fields(0).trim.split(",")(0).substring(1).trim
          val devVersion = fields(1).trim.split(";")(2).trim
    
          val osType = fields(1).trim.split(";")(0).substring(1).trim
          val osVersion = fields(1).trim.split(";")(1).trim
    
          (devType, devVersion, osType, osVersion)
        }
    
        def parseFile = {
          val fileData = sc.textFile("file:///path_to_my_file")
          val filteredData = fileData.map(_.split("\\t")).filter(l => l(5).toLowerCase.matches("recover"))
          filteredData.map(l =>
            flattenTuple(l(0), l(1), l(2), l(3), l(4), l(5), parseClientInfo(l(6)), l(7), l(8), l(9), l(10), l(11), l(12)))
        }
    
        val usefulData = parseFile
        //usefulData.take(1).foreach(println)
        usefulData.persist(StorageLevel.DISK_ONLY)
    
    
        if (!usefulData.isEmpty()) {
    
          /**
            * KPI : RECOVER.CNT
            * AGG_CNT : _._16
            */
          val recover_count = usefulData.map(_._16.toDouble).sum.toInt
          println("recover_count : " + recover_count)
    
          /**
            * KPI : RECOVER.UniqueUser
            * PRS_ID : _._1
            */
          val recover_uu = usefulData.map(l => l._1).distinct.count
          println("recover_uu : " + recover_uu)
    
          val (success, failure) = usefulData.map(l => ((l._1, l._5), l._16)).partitionRDDBy(_._1._2 == "200")
    
          /**
            * KPI : RECOVER.SUCCESS.COUNT
            * AGG_CNTs of SUCCESS : _._2
            */
          val success_cnt = success.map(l => l._2.toDouble).sum.toInt
          println("success_cnt : " + success_cnt)
    
          /**
            * KPI : RECOVER.SUCCESS.UniqueUser
            * PRS_IDs of SUCCESS : _._1._1
            */
          val success_uu = success.map(l => l._1._1).distinct.count.toInt
          println("success_uu : " + success_uu)
    
          /**
            * KPI : RECOVER.FAILURE.COUNT
            * AGG_CNTs of FAILURE : _._2
            */
          val failure_cnt = failure.map(l => l._2.toDouble).sum.toInt
          println("failure_cnt : " + failure_cnt)
    
          /**
            * KPI : RECOVER.FAILURE.UniqueUser
            * PRS_IDs of FAILURE : _._1._1
            */
          val failure_uu = failure.map(l => l._1._1).distinct.count.toInt
          println("failure_uu : " + failure_uu)
    
          /**
            * KPI : RECOVER.FAILSUCCESS.UniqueUser
            * PRS_IDs of SUCCESS and FAILURE : _._1
            */
          val fail_success_uu = success.map(l => l._1).join(failure.map(l => l._1)).count
          println(fail_success_uu)
    
          /**
            * KPI : RECOVER.PCFAILURE.COUNT
            * RESPONSEs : l._5
            * ERROR_CODEs : l._12
            */
          val pc_failure_cnt = usefulData.filter(l => l._5 != "200" && l._12 == "-6015").map(l => l._16.toDouble).sum.toInt
          println("pc_failure_cnt : " + pc_failure_cnt)
    
          /**
            * KPI : RECOVER.PCFAILURE.UniqueUser
            * RESPONSEs : l._5
            * ERROR_CODEs : l._12
            */
          val pc_failure_uu = usefulData.filter(l => l._5 != "200" && l._12 == "-6015").map(l => l._1).distinct.count.toInt
          println("pc_failure_uu : " + pc_failure_uu)
    
          /**
            * KPI : RECOVER.RECORD.COUNT
            * LABELs : l._2
            * PRS_IDs : _._16
            */
          val record_cnt = usefulData.filter(l => l._2.contains("record")).map(l => l._16.toDouble).sum.toInt
          println("record_cnt : " + record_cnt)
    
          /**
            * KPI : RECOVER.RECORD.SUCCESS.UniqueUser
            * RESPONSEs : l._5
            * LABELs : l._2
            * PRS_IDs : _._1
            */
          val record_success_uu = usefulData.filter(l => l._5 == "200" && l._2.contains("record")).map(l => l._1).distinct.count.toInt
          println("record_success_uu : " + record_success_uu)
    
          /**
            * KPI : RECOVER.RECORD.FAILURE.UniqueUser
            * RESPONSEs : l._5
            * LABELs : l._2
            * PRS_IDs : _._1
            */
          val record_failure_uu = usefulData.filter(l => l._5 != "200" && l._2.contains("record")).map(l => l._1).distinct.count.toInt
          println("record_failure_uu : " + record_failure_uu)
    
          /**
            * Group KPI 1 - DVC.VER.UniqueUser
            * DEVICE_TYPEs : l._7
            * PRS_IDs : l._1
            */
          val group_ver_uu = usefulData.map(l => (l._7, l._1)).distinct.aggregateByKey(0)((acc, value) => acc + 1, _ + _)
          println("group_ver_uu : ")
          group_ver_uu.collect().foreach(println)
    
          if (!group_ver_uu.isEmpty) {
    
            /**
              * Group KPI 2 - DVC.UniqueUser
              * DEVICE_TYPE WITHOUT VERSION : l._1
              * COUNT OF SIMILAR DEVICES IRRESPECTIVE OF VERSIONs : l._2
              */
            val group_uu = group_ver_uu.map(l => ((pattern findFirstIn l._1).getOrElse(None), l._2)).aggregateByKey(0)((acc, value) => acc + value, _ + _)
            println("group_uu")
            group_uu.collect.foreach(println)
    
            val (macData, otherDevices) = group_uu.partitionRDDBy(l => l._1.toString.toLowerCase.contains("mac"))
    
            /**
              * KPI : RECOVER.DVC.MAC.UniqueUser
              */
            val mac_uu = macData.aggregate(0)((acc, value) => acc + value._2, _ + _)
            println("mac_uu : " + mac_uu)
    
          }
    
          /**
            * Group KPI 3 - OS.UniqueUser
            * OS_TYPEs : l._9
            * PRS_IDs : l._1
            */
          val group_os_uu = usefulData.map(l => (l._9, l._1)).distinct.aggregateByKey(0)((acc, value) => acc + 1, _ + _)
          println("group_os_uu")
          group_os_uu.collect().foreach(println)
    
          /**
            * Group KPI 2 - OS.VER.UniqueUser
            * OS_TYPEs : l._9
            * OS_VERSIONS : l._10
            * PRS_IDS : l._1
            */
          val group_os_ver_uu = usefulData.map(l => ((l._9, l._10), l._1)).distinct.aggregateByKey(0)((acc, value) => acc + 1, _ + _)
          println("group_os_ver_uu")
          group_os_ver_uu.collect().foreach(println)
        }
      }
    
      implicit class RDDOps[T](rdd: RDD[T]) {
        def partitionRDDBy(f: T => Boolean): (RDD[T], RDD[T]) = {
          val passes = rdd.filter(f)
          val fails = rdd.filter(e => !f(e))
          (passes, fails)
        }
      }
    
      implicit def flattenTuple[A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P](t: (A, B, C, D, E, F, (G, H, I, J), K, L, M, N, O, P)): (A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P) = {
        (t._1, t._2, t._3, t._4, t._5, t._6, t._7._1, t._7._2, t._7._3, t._7._4, t._8, t._9, t._10, t._11, t._12, t._13)
      }
    }

Answer 1

不确定我是否准确地理解了这些问题，但这里是一个尝试：

1. 您可以使用谓词：usefulData.count==0。另一种解决方案是：

   rdd.mapPartitions(iter => Iterator(! iter.hasNext)).reduce(_&&_)

2. 这完全取决于逻辑。基本上，对于每个 KPI，您都在 DAG 中创建了一个新分支。如果您看到重叠的逻辑，您绝对可以将常见的转换保存为多次发生。

3. 对此不确定，我也将 Tuple 用于最多 9-10 个字段。虽然最大尺寸是 22。

4. 几天前我遇到了类似的问题。所以现在是时候问这个问题了：SO Link

5. 当然可以。您的策略应该包括 1) 创建一个新的记录索引列表/数组。 2) 在你的 RDD 上执行转换，最好使用 mapPartions。每个 KPI 的结果应该与输入 RDD 具有相同的基数（使用选项来避免失败）。 3.使用 zip/zipWithIndex KPI RDD。我想这就是你要找的。​​p>

如果您找到剩余答案，请更新/回答此主题。