014 在Spark中完成PV與UV的計算，重在源代碼

1.代碼

object LogPVAndUV{
    def main(args:Array[String]):Unit={
        val conf=new SparkConf()
            .setMaster("local[*]")
            .setAppName("PVAndUV")
        val sc=SparkContext.getOrCreate(conf)
        val logPath="/user/beifeng/spark/logs/page_views.data"
        val logRDD=sc.textFile(logPath)
        val filterRDD=logRDD.filter(_.length>0)
        //轉換
        val mapRDD=filterRDD.map(line=>{
            val arr=line.split("\t")
            if(arr.length==7){
                val date=arr(0).trim
                val url=arr(1)
                val uuid=arr(2)
                (date.subString(0,Math.min(10.date.length)).trim,url,uuid)
            }else{
                (null,null,null)
            }
        }).filter(tuple=>tuple._1!=null&&tuple._1.length>0)
        //PV計算
        val pvRDD=mapRDD
            .filter(tuple=>tuple._2.length>0)
            .map(tuple=>(tuple._1,1))
            .reduceByKey(_+_)
        //UV計算
        val uvRDD=mapRDD
            .filter(tuple=>tuple._3.length>0)
            .map(tuple=>(tuple._1,tuple._3))
            .distinct
            .reduceByKey(_+_)
        //合並
        val pvAndUv=pvRDD.join(uvRDD).map{
            case (date,(pv,uv))=>{
                (date,pv,uv)
            }
        }
        //輸出
        pvAndUv.saveAsTextFile("/user/beifeng/spark/output/"+System.currentTimeMillis())
        sc.stop()
    }
}

 

2.PS

　　rdd.foreachPartition(iter=>{

　　　　//

　　})

　　對iter迭代器中的數據進行輸出，iter表示的是一個分區的所有數據，這里的迭代器和groupbyKey中的實現方式不同，不會產生OOM

　　主要用於將數據輸出到非HDFS的存儲系統中，不如MYSQL,Redis