Spark 1.X 大数据平台

RDD执行transformation和执行action的区别是什么?

1、transformation是得到一个新的RDD，方式很多，比如从数据源生成一个新的RDD，从RDD生成一个新的RDD

2、action是得到一个值，或者一个结果（直接将RDDcache到内存中）。所有的transformation都是采用的懒策略，就是如果只是将transformation提交是不会执行计算的，计算只有在action被提交的时候才被触发。
说明narrow dependency 和 wide dependency的区别? 从计算和容错两方面说明! 在Spark中，每一个 RDD 是对于数据集在某一状态下的表现形式，而这个状态有可能是从前一状态转换而来的，因此换句话说这一个 RDD 有可能与之前的 RDD(s) 有依赖关系。根据依赖关系的不同，可以将
RDD 分成两种不同的类型： Narrow Dependency 和 Wide Dependency 。

Narrow Dependency 指的是 child RDD 只依赖于 parent RDD(s) 固定数量的partition。

Wide Dependency 指的是 child RDD 的每一个partition都依赖于 parent RDD(s) 所有partition。

它们之间的区别可参看下图：

<img src="http://img.blog.csdn.net/20150608143649809" alt="这里写图片描述" title="" style="color: rgb(54, 46, 43); font-family: 'microsoft yahei'; line-height: 26px; background-color: rgb(255, 255, 255); border: none; box-sizing: border-box; max-width: 100%;" /><span style="color: rgb(54, 46, 43); font-family: 'microsoft yahei'; line-height: 26px; background-color: rgb(255, 255, 255);"> </span>

3. RDD cache默认的StorageLevel级别是什么？

1）RDD的cache()方法其实调用的就是persist方法，缓存策略均为MEMORY_ONLY；

2）可以通过persist方法手工设定StorageLevel来满足工程需要的存储级别；

3）cache或者persist并不是action；

4. Spark目前支持哪几种语言的API？

Scala API/Java API/Python API

5. 下载搜狗实验室用户查询日志精版:http://www.sogou.com/labs/dl/q.html (63M),做以下查询:

1) 用户在00:00:00到12:00:00之间的查询数?

package cn.chinahadoop.scala

import org.apache.spark.{SparkContext, SparkConf}

object SogouA {

def main(args: Array[String]) {

if (args.length == 0) {

System.err.println("Usage: SogouA <file1>")

System.exit(1)

}

val conf = new SparkConf().setAppName("SogouA")

val sc = new SparkContext(conf)

val sgRDD=sc.textFile(args(0))

sgRDD.map(_.split('\t')(0)).filter(x => x >= "00:00:00" && x <= "12:00:00").saveAsTextFile(args(1))

sc.stop()

}

}

客户端运行命令：

./spark-submit

--master spark://SparkMaster:7077

--name chinahadoop

--class cn.chinahadoop.scala.SogouA

/home/chinahadoop.jar

hdfs://SparkMaster:9000/data/SogouQ.reduced

hdfs://SparkMaster:9000/data/a

2) 搜索结果排名第一,但是点击次序排在第二的数据有多少?

package cn.chinahadoop.scala

import org.apache.spark.{SparkContext, SparkConf}

object SogouB {

def main(args: Array[String]) {

if (args.length == 0) {

System.err.println("Usage: SogouB <file1>")

System.exit(1)

}

val conf = new SparkConf().setAppName("SogouB")

val sc = new SparkContext(conf)

val sgRDD=sc.textFile(args(0))

println(sgRDD.map(_.split('\t')).filter(_.length ==5).map(_(3).split(' ')).filter(_(0).toInt ==1).filter(_(1).toInt ==2).count)

sc.stop()

}

}

客户端运行命令：与上雷同

3)一个session内查询次数最多的用户的session与相应的查询次数?

package cn.chinahadoop.scala

import org.apache.spark.{SparkContext, SparkConf}

import org.apache.spark.SparkContext._

object SogouC {

def main(args: Array[String]) {

if (args.length == 0) {

System.err.println("Usage: SogouC <file1>")

System.exit(1)

}

val conf = new SparkConf().setAppName("SogouC")

val sc = new SparkContext(conf)

val sgRDD=sc.textFile(args(0))

sgRDD.map(_.split('\t')).filter(_.length ==5).map(x=>(x(1),1)).reduceByKey(_+_).map(x=>(x._2,x._1)).sortByKey(false).map(x=>(x._2,x._1)).take(10).foreach(println)

sc.stop()

}

}

客户端运行命令：与上雷同