sparkstream

image.png

Kafka 分布式的单位是 Partition。如何保证消息有序，需要分几个情况讨论。

• 同一个 Partition 用一个 write ahead log 组织，所以可以保证 FIFO 的顺序。
• 不同 Partition 之间不能保证顺序。但是绝大多数用户都可以通过 message key 来定义，因为同一个 key 的 message 可以保证只发送到同一个 Partition。比如说 key 是 user id，table row id 等等，所以同一个 user 或者同一个 record 的消息永远只会发送到同一个 Partition上，保证了同一个 user 或 record 的顺序。
• 当然，如果你有 key skewness 就有些麻烦，需要特殊处理。

什么是数据倾斜

对 Spark/Hadoop 这样的大数据系统来讲，数据量大并不可怕，可怕的是数据倾斜。数据倾斜指的是，并行处理的数据集中，某一部分（如 Spark 或 Kafka 的一个 Partition）的数据显著多于其它部分，从而使得该部分的处理速度成为整个数据集处理的瓶颈（木桶效应）。

数据倾斜是如何造成的

在 Spark 中，同一个 Stage 的不同 Partition 可以并行处理，而具有依赖关系的不同 Stage 之间是串行处理的。假设某个 Spark Job 分为 Stage 0和 Stage 1两个 Stage，且 Stage 1依赖于 Stage 0，那 Stage 0完全处理结束之前不会处理Stage 1。而 Stage 0可能包含 N 个 Task，这 N 个 Task 可以并行进行

1.具体解决方案

调整并行度分散同一个 Task 的不同 Key: Spark 在做 Shuffle 时，默认使用 HashPartitioner（非 Hash Shuffle ???）对数据进行分区。

2. 自定义Partitioner:

使用自定义的 Partitioner（默认为 HashPartitioner），将原本被分配到同一个 Task 的不同 Key 分配到不同 Task，可以拿上图继续想象一下，通过自定义 Partitioner 可以把原本分到 Task0 的 Key 分到 Task1，那么 Task0 的要处理的数据量就少了。

3. 将 Reduce side（侧） Join 转变为 Map side（侧） Join:

通过 Spark 的 Broadcast 机制，将 Reduce 侧 Join 转化为 Map 侧 Join，避免 Shuffle 从而完全消除 Shuffle 带来的数据倾斜。

4. 为 skew 的 key 增加随机前/后缀:

为数据量特别大的 Key 增加随机前/后缀，使得原来 Key 相同的数据变为 Key 不相同的数据，从而使倾斜的数据集分散到不同的 Task 中，彻底解决数据倾斜问题.

Spark 的 shuffle 过程

Spark shuffle 处于一个宽依赖，可以实现类似混洗的功能，将相同的 Key 分发至同一个 Reducer上进行处理。

Spark有哪些聚合类的算子,我们应该尽量避免什么类型的算子？

在我们的开发过程中，能避免则尽可能避免使用 reduceByKey、join、distinct、repartition 等会进行 shuffle 的算子，尽量使用 map 类的非 shuffle 算子。这样的话，没有 shuffle 操作或者仅有较少 shuffle 操作的 Spark 作业，可以大大减少性能开销。

6Spark为什么快和Hive比较

• 消除了冗余的 HDFS 读写: Hadoop 每次 shuffle 操作后，必须写到磁盘，而 Spark 在 shuffle 后不一定落盘，可以 cache 到内存中，以便迭代时使用。
• 消除了冗余的 MapReduce 阶段: Hadoop 的 shuffle 操作一定连着完整的 MapReduce 操作，冗余繁琐。而 Spark 基于 RDD 提供了丰富的算子操作，且 reduce 操作产生 shuffle 数据，可以缓存在内存中。
• JVM 的优化: Hadoop 每次 MapReduce 操作，启动一个 Task 便会启动一次 JVM，基于进程的操作。而 Spark 每次 MapReduce 操作是基于线程的，只在启动 Executor 是启动一次 JVM，内存的 Task 操作是在线程复用的。每次启动 JVM 的时间可能就需要几秒甚至十几秒，那么当 Task 多了，这个时间 Hadoop 不知道比 Spark 慢了多少。

结论 Spark 快不是绝对的，但是绝大多数，Spark 都比 Hadoop 计算要快。这主要得益于其对 mapreduce 操作的优化以及对 JVM 使用的优化。

Spark Streaming容错体系

Spark操作的数据一般存储在有容错功能的文件系统（比如HDFS、S3）上，从这些系统上的数据生成的RDD也具有容错能力，但是这个不适用于Spark Streaming。为了达到相同的容错能力，通过网络接收到的数据还被复制到其他节点上（默认复制1份，总共2份数据），这就导致错误发生时有两类数据需要恢复。

• 刚收到已经被缓存，但还没有被复制到其他节点的数据。因为没有其他副本，恢复的唯一方法是从数据源重新获取一份。
• 收到了且已经复制到其他节点的数据。这部分数据可以从其他节点恢复。
  此外，我们还要知道有两类可能发生的错误，如下。
• worker节点失效。一旦计算节点失效，所有内存中的数据都会丢失且无法恢复。
• Driver节点失效。如果运行Driver进程的节点失效，那么SparkContext也会随之失效，整个Streaming程序会退出，所有附属的执行节点都会退出，内存中的数据全部丢失。

结果输出容错

结果输出（saveAsTextFiles、foreachRDD等）操作本身提供至少一次级别的容错性能，就是说可能输出多次至外部系统，但可能通过一些辅助手段来实现精准一次的容错效果。
当输出为文件时是可以接受的，因为重复的数据会覆盖前面的数据，结果一致，效果相当于精确一次，其他场景下的输出要想实现精确一次的容错，需要一些额外的操作，有如下两种方法。

• 幂等更新。确保多操作的效果与一次操作的效果相同，比如saveAs***Files即便调用多次，结果还是同一个文件。
• 事务更新。更新时带上事务信息，确保更新只进行一次，比如使用批次时间和RDD的分区编号构造一个事务ID，在更新时使用事务ID来判断是否已经更新，如果已经更新过则跳过，避免重复，实现精准一次的容错效果。

检查点

由于流式计算7×24小时运行的特点，除了考虑具备容错能力，我们还要考虑容错的代价问题。为了避免错误恢复的代价与运行时间成正比增长，Spark提供了检查点功能，用户定期记录中间状态，避免从头开始计算的漫长恢复。
有一情况下必须启用检查点功能，那就是调用了有状态的Transformation操作，比如updateStateByKey或reduceByKeyAndWindow。因为有状态的操作是从程序开始时一直进行的，如果不做检查点，那么计算链接会随着时间一直增长，重新计算的代价也将会是天文数字。

Spark SQL

而且Spark SQL与Apache Hive基本完全兼容，我们可以像使用Hive一样来使用Spark SQL。
使用Spark SQL有两种方式：

• 一种是作为分布式SQL引擎，此时只需要写SQL就可以进行计算，不需要复杂的编码，这也是最受欢迎的一种方式，本节会重点介绍；
• 另外一种模式是在Spark程序中，通过领域API的形式来操作数据（被抽象为DataFrame）。
  Spark SQL在设计上与Hive兼容，是一种“开箱即用”的兼容方式，即可以直接替代Hive的角色，但内部实现完全不同。可以直接使用Hive的meta数据库，语法上也基本完全兼容HiveQL，不同的地方非常少。一般情况下，完全可以使用HiveQL来操作Spark SQL，发现异常时再通过查询手册等方式来解决。
  Spark SQL支持绝大部分的Hive特性，具体如下所示。
  Hive查询语句，包括：
  •　SELECT
  •　GROUP BY
  •　ORDER BY
  •　CLUSTER BY
  •　SORT BY
  所有Hive运算符，包括：
  •　关系运算符（=、⇔、==、＜＞、＜、＞、＞=、＜=等）
  •　数学运算符（+、-、*、/和％等）
  •　逻辑运算符（AND、＆＆、OR、||等）
  •　复杂类型构造器（struct、named_struct）
  •　数学函数（sign、ln、cos等）
  •　字符串函数（instr、length、printf等）
  •　用户自定义函数（UDF）
  •　用户自定义聚合函数（UDAF）
  •　用户自定义序列化格式（SerDes）
  •　求交
  •　JOIN
  •　{LEFT|RIGHT|FULL} OUTER JOIN
  •　LEFT SEMI JOIN
  •　CROSS JOIN
  •　Unions
  •　子查询
  •　SELECT col FROM ( SELECT a + b AS col from t1) t2
  •　Sampling
  •　Explain
  •　表分区
  •　View
  所有Hive DDL函数，包括：
  •　CREATE TABLE
  •　CREATE TABLE AS SELECT
  •　ALTER TABLE
  绝大部分Hive数据类型，包括：
  •　TINYINT
  •　SMALLINT
  •　INT
  •　BIGINT
  •　BOOLEAN
  •　FLOAT
  •　DOUBLE
  •　STRING
  •　BINARY
  •　TIMESTAMP
  •　DATE
  •　ARRAY＜＞
  •　MAP＜＞
  •　STRUCT＜＞
  从Spark 1.4开始，Spark SQL已经开始支持窗口函数了，包括排名函数rank、dense_rank、percent_rank、ntile、row_number，以及分析函数cume_dist、first_value、last_value、lag、lead。使用时，在函数后面要加Over表达式OVER (PARTITION BY …… ORDER BY ……)。

支持的数据类型

Spark SQL和DataFrame支持大部分Hive类型，具体如下所示。
•　数字类型。ByteType、ShortType、IntegerType、LongType、FloatType、DoubleType和DecimalType。
•　字符串类型。StringType。
•　二进制类型。BinaryType。
•　Bool类型。BooleanType。
•　日期时间类型。TimestampType和DateType。
•　复杂类型。ArrayType、MapType和StructType。

DataFrame

Spark SQL除了支持使用SQL方式来查询之外，还提供了编程接口，可以在Spark程序中引用Spark SQL模块。编程时使用的数据抽象是DataFrame，它具有与RDD类似的分布式数据集特点，但还增加了列的概念，这样可以与传统关系型数据库的表对应起来。Spark程序支持的Scala、Java、Python以及R这4种编程语言都可以使用DataFrame。
在Spark中使用DataFrame的过程也很简单，大概需要如下4步：
(1) 初始化环境，一般是创建一个SQLContext对象；
(2) 创建一个DataFrame，可以来源于RDD或其他数据源；
(3) 调用DataFrame操作，是一种领域特定的API，可以实现所有的SQL功能；
(4) 或者通过函数直接执行SQL语句。

DataFrame数据源

DataFrame支持非常多类型的数据源，包括Hive、Avro、Parquet、ORC、JSON、JDBC。而且Spark提供了统一的读写接口。

Catalyst执行优化器

Catalyst是Spark SQL执行优化器的代号，所有Spark SQL语句最终都通过它来解析、优化，最终生成可以执行的Java字节码。因此，Catalyst是Spark SQL最核心的部分。