1. RDD如何持久化数据?
-
有两种方法可以持久存储数据，比如持久存储 persist()和cache() 临时存储在内存中。
-
有不同的存储级别选项，比如MEMORY_ONLY、MEMORY_AND_DISK、DISK_ONLY等等。
-
persist() 和 cache() 使用不同的选项取决于任务的要求。
2. 在Yarn上运行Spark时，是否需要在Yarn cluster的所有节点上都安装Spark ?
-
因为spark运行在Yarn之上，所以它利用Yarn在集群节点上执行命令。
-
所以，你只需要在一个节点上安装Spark。
3. Spark的缺点是什么?
Spark利用了内存。开发人员必须小心。一个普通的开发人员可能会犯以下错误:
-
可能最终在本地节点上运行所有东西，而不是将工作分配到集群。
-
通过使用多个集群，她可能会多次访问某个服务。
4. 什么是RDD？
-
RDD的完整形式是弹性分布式数据集。
-
它是位于网络上的数据的一种表示
-
不可变的——您可以对rdd进行操作以生成另一个rdd，但是您不能更改它。
-
分区/并行——位于RDD上的数据是并行操作的。RDD上的任何操作都是使用多个节点完成的。
-
弹性——如果一个承载分区的节点发生故障，另一个节点将获取它的数据。
-
RDD提供两种操作: 转换和操作。
5. 转换是什么?
-
转换是应用于RDD(弹性分布式数据集)的函数。转换会导致另一个RDD。
-
直到一个动作发生，转换才会被执行。
-
转换的例子有:
-
map()——在RDD的每个元素上应用传递给它的函数，生成一个新的RDD。
-
filter()——通过从传递函数参数的当前RDD中选择元素来创建一个新的RDD。
6. Action是什么?
-
一个Action将数据从RDD带回本地机器。
-
一个Action的执行会导致之前创建的所有转换。
-
Action的例子是:
-
reduce()——执行一次又一次传递的函数，直到只剩下一个值。函数应该有两个参数并返回一个值。
-
take()——将所有值从RDD返回到本地节点。
7. 假设我在RDD中有一个庞大的数字列表(假设是myrdd)。我写了以下代码来计算平均值:
def myAvg(x, y):
return (x+y)/2.0;
avg = myrdd.reduce (myAvg);
它有什么问题? 你会怎么改正呢?
-
简单地求和，然后除以计数。
- def sum(x, y):
- return x+y;
- total = myrdd.reduce(sum);
- avg = total / myrdd.count();
-
上述代码的唯一问题是，总数可能会变得非常大，从而导致流溢出。所以，我宁愿每一个数除以计数，然后按照下面的方式求和。
- cnt = myrdd.count();
- def devideByCnd(x):
- return x/cnt;
- myrdd1 = myrdd.map(devideByCnd);
- avg = myrdd.reduce(sum);
8. 假设在HDFS文件中有一个巨大的数字列表。每一行都有一个数字。我要计算这些数的平方和的平方根。你会怎么做?
- #我们将首先从HDFS在spark上以RDD的形式加载文件
- numsAsText = sc.textFile (“hdfs: / /hadoop1.knowbigdata.com/user/student/sgiri/mynumbersfile.txt”);
- #定义用来计算squaresdef toSqInt(str)的函数:
- v = int(str);
- return v*v;
- #在spark rdd上作为转换运行函数
- nums = numsAsText.map (toSqInt);
- #运行sum as reduce操作
- total= nums.reduce(sum)
- 最后计算平方根。我们需要导入math。
- import math;
- print math.sqrt(total);
9. 下面的方法正确吗? sqrtOfSumOfSq是一个有效的减速器吗?
- numsAsText =sc.textFile(“hdfs://hadoop1.knowbigdata.com/user/student/sgiri/mynumbersfile.txt”);
- def toInt(str):
- return int(str);
- nums = numsAsText.map(toInt);
- def sqrtOfSumOfSq(x, y):
- return math.sqrt(x*x+y*y);
- total = nums.reduce(sqrtOfSumOfSq)
- import math;
- print math.sqrt(total);
- 是的。该方法是正确的，sqrtOfSumOfSq是一个有效的减速器。
10. 你能比较一下9 和10方法的优缺点吗？
- 9在map()中进行平方运算并在 action方法中对reduce求和
- 10在执行reduce操作时做平方运算。
- 9方法将会更快，因为10中，由于调用math.sqrt()，所以减速器代码很重，并且减速器代码通常大约执行n-1次spark RDD。
- 方法9的唯一缺点是有一个巨大的整数溢出的机会，因为我是作为map的一部分计算平方和的。
- 原则：能在map 处做的操作尽量在 map 处做。
11. 如果你要计算spark上每个单词的总数，你会怎么做?
- 这将在spark中加载bigtextfile.txt作为RDD
- sc.textFile (“hdfs: / /hadoop1.knowbigdata.com/user/student/sgiri/bigtextfile.txt”);
- 定义一个函数，可以将每一行分解成单词
- def toWords(line):
- return line.split();
- 在spark上的RDD的每个元素上运行toWords函数作为平面映射转换。
- 我们将使用flatMap而不是map，因为我们的函数返回了多个值。
- words= lines.flatMap(toWords);
- 将每个单词转换成(键，值)对。她的键将是单词本身，值将是1。
- def toTuple(word):
- return (word, 1);
- wordsTuple = words.map(toTuple);
- 现在我们可以轻松地执行reduceByKey()操作。
- def sum(x, y):
- return x+y;
- counts = wordsTuple.reduceByKey(sum)
- #现在,打印
- counts.collect ()
12. 在一个非常大的文本文件中，您只想检查是否存在特定的关键字。你会如何使用Spark?
- lines = sc.textFile(“hdfs://hadoop1.knowbigdata.com/user/student/sgiri/bigtextfile.txt”);
- def isFound(line):
- if line.find(“mykeyword”) > -1:
- return 1;
- return 0;
- foundBits = lines.map(isFound);
- sum = foundBits.reduce(sum);
- if sum > 0:
- print “FOUND”;
- else:
- print “NOT FOUND”;
13. 你能在之前的答案中改进这段代码的性能吗?
-
是的。即使我们寻找的单词已经找到，搜索也没有停止。我们的map代码将在所有节点上继续执行，这是非常低效的。
-
我们可以使用累加器来报告单词是否已经找到，然后停止工作.