大数据Spark “蘑菇云”行动第65课： 页面跳转功能代码骨架分析与实战 放scala代码

大数据Spark “蘑菇云”行动第65课： 页面跳转功能代码骨架分析与实战

 

页面调转类似图计算

Spark GraphX中的二跳邻算法
用户在页面上产生的首次点击被称为“二跳”，二跳的次数即为“二跳量”。二跳量与浏览量的比值称为页面的二跳率



 

 



 



 

 



 

 



 

 



 

 



 



 

 



 

 



 



 

 



 

 

 

PageJumpOnce.scala

package com.dt.spak.SparkApps.pageLinks

import org.apache.spark.SparkConf

import org.apache.spark.SparkContext

object PageJumpOnce {

  def main(args: Array[String]): Unit = {

    val sparkConf = new SparkConf().setAppName("PageJumpOnce")

      .setMaster("spark://192.168.189.1:7077").setJars(List(

        // .setMaster("local[5]").setJars(List(

        "/usr/local/spark-1.6.1-bin-hadoop2.6/lib/mysql-connector-java-5.1.13-bin.jar",

        "/usr/local/IMF_testdata/PageJumpOnce.jar"))

    val sc = new SparkContext(sparkConf)

    val data = PageUtils.getData(sc)

    val resolvedTask = PageUtils.parseArgsFromUser(args)

    val aggregated = PageUtils.aggerateOnSession(data, resolvedTask)

    val pageSplited = PageUtils.aggeratePageSplit(aggregated)

    PageUtils.getResult(pageSplited, resolvedTask).foreachPartition { partition =>

      {

      }

    }

  }

}

 

 

PageUtils.scala

package com.dt.spak.SparkApps.pageLinks

import org.apache.spark.SparkContext

import org.apache.spark.rdd.RDD

object PageUtils {

  def getData(sc: SparkContext): RDD[String] = {

    generateData(sc)

  }

  def generateData(sc: SparkContext): RDD[String] = {

    /**

     * 数据可以来自于流处理系统，也可以来自批处理系统

     *

     */

    // MapPartitionsRDD[(String,String)]  //要是同一个包  org.apache.spark.rdd

    sc.parallelize("I into Spark so much".split(" "))

  }

  def parseArgsFromUser(args: Array[String]): Long = {

    0L

  }

  def aggerateOnSession(rdd: RDD[String], taskType: Long): RDD[String] = {

    rdd

  }

  def aggeratePageSplit(rdd: RDD[String]): RDD[String] = {

    rdd

  }

  def getResult(rdd: RDD[String], taskType: Long): RDD[String] = {

    rdd

  }

}