1. spark如何实现容错性?
spark的容错性有2个手段, 检查点和RDD血统
- 检查点: checkpoint, 将RDD数据持久化到hdfs中, 并打断rdd的linage
- RDD血统: 每个RDD都记录有自己的血统, 一个计算序列, 一旦有哪个RDD分区丢失或其它问题, 可以根据自身的RDD重新计算,
重算可以在某个几点分区下执行, 不会导致系统回滚
2. spark和hadoop的mr有什么区别
首先, 两者都是基于MR的计算模型, 但有很多不同
- hadoop的一个job, 只有map操作和reduce操作, 多数据的表达能力欠缺, 每次reduce后要将数据存到hdfs上; 为了完成一个目标往往要串行执行多个job, 造成大量的HDFS IO浪费
- spark将1个job划分为多个stage, stage的划分是通过RDD之间的依赖关系判定的; 每个stage产生的结果会存在内存中, 避免多个stage串行的情况下对hdfs的读写请求. 而且spark拥有很多算子, 数据表现力上很强
3. RDD中的宽窄依赖? 有什么作用?
-
窄依赖是指父RDD的每个分区都只被子RDD的一个或几个分区所使用(不是依赖所有的父RDD分区)。
宽依赖就是指父RDD的分区被多个子RDD的分区所依赖。
例如，map就是一种窄依赖，而join则会导致宽依赖，主要是看有没有shuffle操作。 - 宽窄依赖的作用是用来划分stage。
- 窄依赖分为:
-
OneToOneDependency: 子RDD的一个分区只依赖与父RDD的一个分区 -
RangeDependency: 子RDD的一个分区依赖父RDD的多个分区, 这种依赖只在UnionRDD所使用。
-
4. cache和persist的区别？
它们都是用来进行缓存RDD的。
- cache是特定的persist，rdd中cache的缓存级别是MEMORY_ONLY，cache调用了persist；
- persist可以设置不同的缓存级别。
- cache和persist, checkponit都是懒计算的, 需要执行action算子时才真正触发执行动作;
因此需要在cache(), checkpint()后面加上类似foreach(_ => Unit)的算子进行触发
一共有几种缓存级别
- MEMORY_ONLY：数据保存在内存中，如果内存不够，数据可能就不会持久化；
- MEMORY_AND_DISK：数据优先保存在内存中，如果内存不够则会存到磁盘中；
- MEMORY_ONLY_SER：和MEMORY_ONLY类似，区别是会将RDD中的数据进行序列化，这种方式更加节省内存；
- MEMORY_AND_DISK_SER：和MEMORY_AND_DISK类似，区别是会将RDD中的数据进行序列化，这种方式更加节省内存；
- DISK_ONLY：将数据全部写入磁盘文件中；
- MEMORY_ONLY_2, MEMORY_AND_DISK_2, 等等：这种有后缀_2的，代表的是将每个持久化的数据，都复制一份副本，并将副本保存到其他节点上。这种基于副本的持久化机制主要用于进行容错。
5. 广播变量和累加器?
-
广播变量
广播变量是把一个集合, 通过driver节点发给每个executor, executor只能使用这个集合但不能修改这个集合. 使用广播变量时, driver要先把一个RDD进行collect(), 然后再广播发送出去, 因此, driver的网络压力就会很大 -
累加器
累加器是在Driver端记录的一个全局值, 在driver声明, 在executor端更新, 且只能触发add()方法, 最终由driver端调用collect得到最终结果
6. groupByKey和reduceByKey哪个好
reduceByKey更好
- reduceByKey中指定的聚合函数, 会在map端先进性聚合, 再发往reducer进行计算;
- 而groupByKey直接将数据发到reducer, 网络通信量会大很多
7. kafka
-
Kafka 分布式的情况下，如何保证消息的顺序?
kafka分布式情况下, 数据只能保证局部有序, 不能保证全局有序. 什么意思?- kafka是划分partition的, 一个partition因为被一个write agead log记录, 因此同一个partition下的数据可以有序;
- 不同partition之间, 不能保证顺序。
不过绝大多数用户都可以通过 message key 来定义，因为同一个 key 的 message 可以保证只发送到同一个 Partition。
比如说 key 是 user id，table row id 等等，所以同一个 user 或者同一个 record 的消息永远只会发送到同一个 Partition上，保证了同一个 user 或 record 的顺序。 - SparkStreaming可以设置手动提交offset, 已处理过的offset可以保存在zookeeper中
-
Spark streaming 读取kafka数据的两种方式?
- 基于Receiver方式
需要使用单独的Receiver线程来异步获取Kafka数据。Spark Streaming启动时，会在Executor中同时启动Receiver异步线程用于从Kafka持续获取数据，获取的数据先存储在Receiver中(存储方式由StorageLevel决定)，后续，当Batch Job触发后，这些数据会被转移到剩下的Executor中被处理。处理完毕后，Receiver会自动更新Zookeeper中的Offset。 - 基于Direct(No Receiver)方式
不需要使用单独的Receiver线程从Kafka获取数据。Spark Streaming Batch Job触发时，Driver端确定要读取的Topic-Partition的OffsetRange，然后由Executor并行从Kafka各Partition读取数据并计算。
- 基于Receiver方式
-
如果Spark Streaming停掉了，如何保证Kafka的重新运作是合理的呢
这主要依赖Spark Streaming的快速故障恢复机制.- 传统流处理系统会在其他节点上重启失败的连续算子，这种方法需要等待转移到的节点先处理完之前的任务才能继续执行新任务;
- 但在Spark Streaming中, 这个节点的任务将均匀地分散到集群中的节点进行计算，相对于传递故障恢复机制能够更快地恢复。
-
说说Spark Streaming 是如何收集和处理数据的，一句话?
在 Spark Streaming 中，数据采集是逐条进行的，而数据处理是按批 mini batch进行的，因此 Spark Streaming 会先设置好批处理间隔 batch duration，当超过批处理间隔就会把采集到的数据汇总起来成为一批数据交给系统去处理。
8. Spark处理数据倾斜有什么好方法
简单一句: Spark 数据倾斜的几种场景以及对应的解决方案，包括避免数据源倾斜，调整并行度，使用自定义 Partitioner，使用 Map 侧 Join 代替 Reduce 侧 Join（内存表合并），给倾斜 Key 加上随机前缀等。下面分情况讨论:
-
如果是多个不同的key落在一个分区导致的数据倾斜
- 可以调大并行度:
spark.default.parallelism : rdd分区数 spark.sql.shuffle.partitions: SparkSQL通过通过这个值修改并发数。- 或者可以自定义Partitioner
spark比hive快的原因主要有2, 一个是shuffle后的结果可以存储在磁盘中, 另一个是DAG执行计划里, 多个job串联时结果不必回写到HDFS中
-
如果是join产生的数据倾斜, 且发现其中有一个表很小
- 可以将小表通过broadcast广播出去
这样完全避免了shuffle, 让join完全变成maper端的join - 或者可以把大表的key增加随机前缀, 小表扩大n倍进行join, 最后再删除key的前缀即可
- 可以将小表通过broadcast广播出去
-
如果是1个非常非常大的key导致的数据倾斜
- 可以将这个key拿出来单独处理, 给这个key加上随机前缀, 打散这个key
- 或者把这些倾斜的key过滤出来后的RDD或DataFrame广播出去, 让其进行map端的聚合
9. Spark为什么快，Spark SQL 一定比 Hive 快吗
spark比hive快的原因主要有2, 一个是shuffle后的结果可以存储在磁盘中, 另一个是DAG执行计划里, 多个job串联时结果不必回写到HDFS中.
因此, spark是否比hive快要是有一定条件的, 一个反例是: 当查询只有1次shuffle操作的时候, hive往往比spark还快, 因为MR的map任务往往生成的更多
Select month_id, sum(sales) from T group by month_id; --反例
10. Job 和 Task 怎么理解
-
Job
Job来源于每次的action算子, 每用action算子获取一次结果就会生成一个Job, Job提交给DAGScheduler后, 会分解成task和stage -
Task
Task是在每个Executor的CPU上执行的任务, Task 分为 ShuffleMapTask 和 ResultTask 两种- ShuffleMapStage 中的 Task 为 ShuffleMapTask
- ResultStage 中的 Task 为 ResultTask
11. Spark作业提交-执行流程是怎么样的
-
spark-submit提交代码后，执行new SparkContext()，在SparkContext里构造DAGScheduler和TaskScheduler。 -
DAGScheduler为1个Job生成多个Stage(stage分为ShuffleMapStage和ResultStage), 每个Stage创建一个TaskSet. -
TaskScheduler会把每一个TaskSet里的Task，提交到Executor上执行。 - Executor上有线程池(CachedTHreadPool), 每收到一个Task. 先封装成RunnerTask, 再从线程池中取出一个线程执行这个Task
12. Spark UDF?
Spark SQL UDF 其实是一个Scala函数，被catalyst封装成一个 Expression 结点，最后通过 eval 方法计根据当前 Row 计算 UDF 的结果。
UDF 对表中的单行进行转换，以便为每行生成单个对应的输出值。
-
执行步骤:
- 定义函数
- 将函数注册为UDF
以下示例代码使用 SQL 别名为 CTOF 来注册我们的转换 UDF，然后在 SQL 查询使用它来转换每个城市的温度。
- 将函数注册为UDF
val df = sqlContext.read.json("temperatures.json") df.registerTempTable("citytemps") // Register the UDF with our SQLContext sqlContext.udf.register("CTOF", (degreesCelcius: Double) => ((degreesCelcius * 9.0 / 5.0) + 32.0)) sqlContext.sql("SELECT city, CTOF(avgLow) AS avgLowF, CTOF(avgHigh) AS avgHighF FROM citytemps").show() Spark SQL定义了
UDF1到UDF22共22个类，UDF 最多支持22个输入参数。
13. Spark UDAF?
- 用户定义的聚合函数（User-defined aggregate functions, UDAF）同时处理多行，并且返回一个结果，通常结合使用 GROUP BY 语句（例如 COUNT 或 SUM）。为了简单起见，我们将实现一个叫 SUMPRODUCT 的 UDAF 来计算以库存来分组的所有车辆零售价值
- 使用方法: 通过扩展
UserDefinedAggregateFunction类使用
下面例子中我们定义了一个名为 SumProductAggregateFunction 的类，并且为它取了一个名为 SUMPRODUCT 的别名- 输入文件格式
{"Make":"Honda","Model":"Pilot","RetailValue":32145.0,"Stock":4} {"Make":"Honda","Model":"Civic","RetailValue":19575.0,"Stock":11} {"Make":"Honda","Model":"Ridgeline","RetailValue":42870.0,"Stock":2} {"Make":"Jeep","Model":"Cherokee","RetailValue":23595.0,"Stock":13} {"Make":"Jeep","Model":"Wrangler","RetailValue":27895.0,"Stock":4} {"Make":"Volkswagen","Model":"Passat","RetailValue":22440.0,"Stock":2}- 使用UDAF
// 定义SparkSQL UDAF逻辑 private class SumProductAggregateFunction extends UserDefinedAggregateFunction { // 定义UDF输入schema: (Double price, Long quantity) def inputSchema: StructType = new StructType().add("price", DoubleType).add("quantity", LongType) // 定义输出schema: Output = (Double total) def bufferSchema: StructType = new StructType().add("total", DoubleType) def dataType: DataType = DoubleType def deterministic: Boolean = true // true: our UDAF's output given an input is deterministic def initialize(buffer: MutableAggregationBuffer): Unit = { buffer.update(0, 0.0) // 初始化 result to 0.0 } def update(buffer: MutableAggregationBuffer, input: Row): Unit = { val sum = buffer.getDouble(0) // Intermediate result to be updated val price = input.getDouble(0) // First input parameter val qty = input.getLong(1) // Second input parameter buffer.update(0, sum + (price * qty)) // Update the intermediate result } // Merge intermediate result sums by adding them def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = { buffer1.update(0, buffer1.getDouble(0) + buffer2.getDouble(0)) } // THe final result will be contained in 'buffer' def evaluate(buffer: Row): Any = { buffer.getDouble(0) } } def main (args: Array[String]) { val conf = new SparkConf().setAppName("Scala UDAF Example") val sc = new SparkContext(conf) val sqlContext = new SQLContext(sc) val testDF = sqlContext.read.json("inventory.json") testDF.registerTempTable("inventory") // 注册UDAF到SQLContext sqlContext.udf.register("SUMPRODUCT", new SumProductAggregateFunction) sqlContext.sql("SELECT Make, SUMPRODUCT(RetailValue,Stock) AS InventoryValuePerMake FROM inventory GROUP BY Make").show() }
