first commit
This commit is contained in:
张乾
@ -0,0 +1,226 @@
|
||||
|
||||
|
||||
因收到Google相关通知，网站将会择期关闭。相关通知内容
|
||||
|
||||
|
||||
10 广播变量 & 累加器：共享变量是用来做什么的？
|
||||
你好，我是吴磊。
|
||||
|
||||
今天是国庆第一天，首先祝你节日快乐。专栏上线以来，有不少同学留言说期待后续内容，所以国庆期间我们仍旧更新正文内容，让我们一起把基础知识模块收个尾。
|
||||
|
||||
学习过RDD常用算子之后，回顾这些算子，你会发现它们都是作用（Apply）在RDD之上的。RDD的计算以数据分区为粒度，依照算子的逻辑，Executors以相互独立的方式，完成不同数据分区的计算与转换。
|
||||
|
||||
不难发现，对于Executors来说，分区中的数据都是局部数据。换句话说，在同一时刻，隶属于某个Executor的数据分区，对于其他Executors来说是不可见的。
|
||||
|
||||
不过，在做应用开发的时候，总会有一些计算逻辑需要访问“全局变量”，比如说全局计数器，而这些全局变量在任意时刻对所有的Executors都是可见的、共享的。那么问题来了，像这样的全局变量，或者说共享变量，Spark又是如何支持的呢？
|
||||
|
||||
今天这一讲，我就来和你聊聊Spark共享变量。按照创建与使用方式的不同，Spark提供了两类共享变量，分别是广播变量（Broadcast variables）和累加器（Accumulators）。接下来，我们就正式进入今天的学习，去深入了解这两种共享变量的用法、以及它们各自的适用场景。
|
||||
|
||||
广播变量（Broadcast variables）
|
||||
|
||||
我们先来说说广播变量。广播变量的用法很简单，给定普通变量x，通过调用SparkContext下的broadcast API即可完成广播变量的创建，我们结合代码例子看一下。
|
||||
|
||||
val list: List[String] = List("Apache", "Spark")
|
||||
|
||||
// sc为SparkContext实例
|
||||
val bc = sc.broadcast(list)
|
||||
|
||||
|
||||
在上面的代码示例中，我们先是定义了一个字符串列表list，它包含“Apache”和“Spark”这两个单词。然后，我们使用broadcast函数来创建广播变量bc，bc封装的内容就是list列表。
|
||||
|
||||
// 读取广播变量内容
|
||||
bc.value
|
||||
// List[String] = List(Apache, Spark)
|
||||
|
||||
// 直接读取列表内容
|
||||
list
|
||||
// List[String] = List(Apache, Spark)
|
||||
|
||||
|
||||
使用broadcast API创建广播变量
|
||||
|
||||
广播变量创建好之后，通过调用它的value函数，我们就可以访问它所封装的数据内容。可以看到调用bc.value的效果，这与直接访问字符串列表list的效果是完全一致的。
|
||||
|
||||
看到这里，你可能会问：“明明通过访问list变量就可以直接获取字符串列表，为什么还要绕个大弯儿，先去封装广播变量，然后又通过它的value函数来获取同样的数据内容呢？”实际上，这是个非常好的问题，要回答这个问题，咱们需要做个推演，看看直接访问list变量会产生哪些弊端。
|
||||
|
||||
在前面的几讲中，我们换着花样地变更Word Count的计算逻辑。尽管Word Count都快被我们“玩坏了”，不过，一以贯之地沿用同一个实例，有助于我们通过对比迅速掌握新的知识点、技能点。因此，为了让你迅速掌握广播变量的“精髓”，咱们不妨“故技重施”，继续在Word Count这个实例上做文章。
|
||||
|
||||
普通变量的痛点
|
||||
|
||||
这一次，为了对比使用广播变量前后的差异，我们把Word Count变更为“定向计数”。
|
||||
|
||||
所谓定向计数，它指的是只对某些单词进行计数，例如，给定单词列表list，我们只对文件wikiOfSpark.txt当中的“Apache”和“Spark”这两个单词做计数，其他单词我们可以忽略。结合[第1讲]Word Count的完整代码，这样的计算逻辑很容易实现，如下表所示。
|
||||
|
||||
import org.apache.spark.rdd.RDD
|
||||
val rootPath: String = _
|
||||
val file: String = s"${rootPath}/wikiOfSpark.txt"
|
||||
// 读取文件内容
|
||||
val lineRDD: RDD[String] = spark.sparkContext.textFile(file)
|
||||
// 以行为单位做分词
|
||||
val wordRDD: RDD[String] = lineRDD.flatMap(line => line.split(" "))
|
||||
|
||||
// 创建单词列表list
|
||||
val list: List[String] = List("Apache", "Spark")
|
||||
// 使用list列表对RDD进行过滤
|
||||
val cleanWordRDD: RDD[String] = wordRDD.filter(word => list.contains(word))
|
||||
// 把RDD元素转换为（Key，Value）的形式
|
||||
val kvRDD: RDD[(String, Int)] = cleanWordRDD.map(word => (word, 1))
|
||||
// 按照单词做分组计数
|
||||
val wordCounts: RDD[(String, Int)] = kvRDD.reduceByKey((x, y) => x + y)
|
||||
// 获取计算结果
|
||||
wordCounts.collect
|
||||
// Array[(String, Int)] = Array((Apache,34), (Spark,63))
|
||||
|
||||
|
||||
将上述代码丢进spark-shell，我们很快就能算出，在wikiOfSpark.txt文件中，“Apache”这个单词出现了34次，而“Spark”则出现了63次。虽说得出计算结果挺容易的，不过知其然，还要知其所以然，接下来，咱们一起来分析一下，这段代码在运行时是如何工作的。
|
||||
|
||||
|
||||
|
||||
如上图所示，list变量本身是在Driver端创建的，它并不是分布式数据集（如lineRDD、wordRDD）的一部分。因此，在分布式计算的过程中，Spark需要把list变量分发给每一个分布式任务（Task），从而对不同数据分区的内容进行过滤。
|
||||
|
||||
在这种工作机制下，如果RDD并行度较高、或是变量的尺寸较大，那么重复的内容分发就会引入大量的网络开销与存储开销，而这些开销会大幅削弱作业的执行性能。为什么这么说呢？
|
||||
|
||||
要知道，Driver端变量的分发是以Task为粒度的，系统中有多少个Task，变量就需要在网络中分发多少次。更要命的是，每个Task接收到变量之后，都需要把它暂存到内存，以备后续过滤之用。换句话说，在同一个Executor内部，多个不同的Task多次重复地缓存了同样的内容拷贝，毫无疑问，这对宝贵的内存资源是一种巨大的浪费。
|
||||
|
||||
RDD并行度较高，意味着RDD的数据分区数量较多，而Task数量与分区数相一致，这就代表系统中有大量的分布式任务需要执行。如果变量本身尺寸较大，大量分布式任务引入的网络开销与内存开销会进一步升级。在工业级应用中，RDD的并行度往往在千、万这个量级，在这种情况下，诸如list这样的变量会在网络中分发成千上万次，作业整体的执行效率自然会很差 。
|
||||
|
||||
面对这样的窘境，我们有没有什么办法，能够避免同一个变量的重复分发与存储呢？答案当然是肯定的，这个时候，我们就可以祭出广播变量这个“杀手锏”。
|
||||
|
||||
广播变量的优势
|
||||
|
||||
想要知道广播变量到底有啥优势，我们可以先用广播变量重写一下前面的代码实现，然后再做个对比，很容易就能发现广播变量为什么能解决普通变量的痛点。
|
||||
|
||||
import org.apache.spark.rdd.RDD
|
||||
val rootPath: String = _
|
||||
val file: String = s"${rootPath}/wikiOfSpark.txt"
|
||||
// 读取文件内容
|
||||
val lineRDD: RDD[String] = spark.sparkContext.textFile(file)
|
||||
// 以行为单位做分词
|
||||
val wordRDD: RDD[String] = lineRDD.flatMap(line => line.split(" "))
|
||||
|
||||
// 创建单词列表list
|
||||
val list: List[String] = List("Apache", "Spark")
|
||||
// 创建广播变量bc
|
||||
val bc = sc.broadcast(list)
|
||||
// 使用bc.value对RDD进行过滤
|
||||
val cleanWordRDD: RDD[String] = wordRDD.filter(word => bc.value.contains(word))
|
||||
// 把RDD元素转换为（Key，Value）的形式
|
||||
val kvRDD: RDD[(String, Int)] = cleanWordRDD.map(word => (word, 1))
|
||||
// 按照单词做分组计数
|
||||
val wordCounts: RDD[(String, Int)] = kvRDD.reduceByKey((x, y) => x + y)
|
||||
// 获取计算结果
|
||||
wordCounts.collect
|
||||
// Array[(String, Int)] = Array((Apache,34), (Spark,63))
|
||||
|
||||
|
||||
可以看到，代码的修改非常简单，我们先是使用broadcast函数来封装list变量，然后在RDD过滤的时候调用bc.value来访问list变量内容。你可不要小看这个改写，尽管代码的改动微乎其微，几乎可以忽略不计，但在运行时，整个计算过程却发生了翻天覆地的变化。
|
||||
|
||||
|
||||
|
||||
在使用广播变量之前，list变量的分发是以Task为粒度的，而在使用广播变量之后，变量分发的粒度变成了以Executors为单位，同一个Executor内多个不同的Tasks只需访问同一份数据拷贝即可。换句话说，变量在网络中分发与存储的次数，从RDD的分区数量，锐减到了集群中Executors的个数。
|
||||
|
||||
要知道，在工业级系统中，Executors个数与RDD并行度相比，二者之间通常会相差至少两个数量级。在这样的量级下，广播变量节省的网络与内存开销会变得非常可观，省去了这些开销，对作业的执行性能自然大有裨益。
|
||||
|
||||
好啦，到现在为止，我们讲解了广播变量的用法、工作原理，以及它的优势所在。在日常的开发工作中，当你遇到需要多个Task共享同一个大型变量（如列表、数组、映射等数据结构）的时候，就可以考虑使用广播变量来优化你的Spark作业。接下来，我们继续来说说Spark支持的第二种共享变量：累加器。
|
||||
|
||||
累加器（Accumulators）
|
||||
|
||||
累加器，顾名思义，它的主要作用是全局计数（Global counter）。与单机系统不同，在分布式系统中，我们不能依赖简单的普通变量来完成全局计数，而是必须依赖像累加器这种特殊的数据结构才能达到目的。
|
||||
|
||||
与广播变量类似，累加器也是在Driver端定义的，但它的更新是通过在RDD算子中调用add函数完成的。在应用执行完毕之后，开发者在Driver端调用累加器的value函数，就能获取全局计数结果。按照惯例，咱们还是通过代码来熟悉累加器的用法。
|
||||
|
||||
聪明的你可能已经猜到了，我们又要对Word Count“动手脚”了。在第1讲的Word Count中，我们过滤掉了空字符串，然后对文件wikiOfSpark.txt中所有的单词做统计计数。
|
||||
|
||||
不过这一次，我们在过滤掉空字符的同时，还想知道文件中到底有多少个空字符串，这样我们对文件中的“脏数据”就能做到心中有数了。
|
||||
|
||||
注意，这里对于空字符串的计数，不是主代码逻辑，它的计算结果不会写入到Word Count最终的统计结果。所以，只是简单地去掉filter环节，是无法实现空字符计数的。
|
||||
|
||||
那么，你自然会问：“不把filter环节去掉，怎么对空字符串做统计呢？”别着急，这样的计算需求，正是累加器可以施展拳脚的地方。你可以先扫一眼下表的代码实现，然后我们再一起来熟悉累加器的用法。
|
||||
|
||||
import org.apache.spark.rdd.RDD
|
||||
val rootPath: String = _
|
||||
val file: String = s"${rootPath}/wikiOfSpark.txt"
|
||||
// 读取文件内容
|
||||
val lineRDD: RDD[String] = spark.sparkContext.textFile(file)
|
||||
// 以行为单位做分词
|
||||
val wordRDD: RDD[String] = lineRDD.flatMap(line => line.split(" "))
|
||||
|
||||
// 定义Long类型的累加器
|
||||
val ac = sc.longAccumulator("Empty string")
|
||||
|
||||
// 定义filter算子的判定函数f，注意，f的返回类型必须是Boolean
|
||||
def f(x: String): Boolean = {
|
||||
if(x.equals("")) {
|
||||
// 当遇到空字符串时，累加器加1
|
||||
ac.add(1)
|
||||
return false
|
||||
} else {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
// 使用f对RDD进行过滤
|
||||
val cleanWordRDD: RDD[String] = wordRDD.filter(f)
|
||||
// 把RDD元素转换为（Key，Value）的形式
|
||||
val kvRDD: RDD[(String, Int)] = cleanWordRDD.map(word => (word, 1))
|
||||
// 按照单词做分组计数
|
||||
val wordCounts: RDD[(String, Int)] = kvRDD.reduceByKey((x, y) => x + y)
|
||||
// 收集计数结果
|
||||
wordCounts.collect
|
||||
|
||||
// 作业执行完毕，通过调用value获取累加器结果
|
||||
ac.value
|
||||
// Long = 79
|
||||
|
||||
|
||||
与第1讲的Word Count相比，这里的代码主要有4处改动：
|
||||
|
||||
|
||||
使用SparkContext下的longAccumulator来定义Long类型的累加器；
|
||||
定义filter算子的判定函数f，当遇到空字符串时，调用add函数为累加器计数；
|
||||
以函数f为参数，调用filter算子对RDD进行过滤；
|
||||
作业完成后，调用累加器的value函数，获取全局计数结果。
|
||||
|
||||
|
||||
你不妨把上面的代码敲入到spark-shell里，直观体验下累加器的用法与效果，ac.value给出的结果是79，这说明以空格作为分隔符切割源文件wikiOfSpark.txt之后，就会留下79个空字符串。
|
||||
|
||||
另外，你还可以验证wordCounts这个RDD，它包含所有单词的计数结果，不过，你会发现它的元素并不包含空字符串，这与我们预期的计算逻辑是一致的。
|
||||
|
||||
除了上面代码中用到的longAccumulator，SparkContext还提供了doubleAccumulator和collectionAccumulator这两种不同类型的累加器，用于满足不同场景下的计算需要，感兴趣的话你不妨自己动手亲自尝试一下。
|
||||
|
||||
其中，doubleAccumulator用于对Double类型的数值做全局计数；而collectionAccumulator允许开发者定义集合类型的累加器，相比数值类型，集合类型可以为业务逻辑的实现，提供更多的灵活性和更大的自由度。
|
||||
|
||||
不过，就这3种累加器来说，尽管类型不同，但它们的用法是完全一致的。都是先定义累加器变量，然后在RDD算子中调用add函数，从而更新累加器状态，最后通过调用value函数来获取累加器的最终结果。
|
||||
|
||||
好啦，到这里，关于累加器的用法，我们就讲完了。在日常的开发中，当你遇到需要做全局计数的场景时，别忘了用上累加器这个实用工具。
|
||||
|
||||
重点回顾
|
||||
|
||||
今天的内容讲完了，我们一起来做个总结。今天这一讲，我们重点讲解了广播变量与累加器的用法与适用场景。
|
||||
|
||||
广播变量由Driver端定义并初始化，各个Executors以只读（Read only）的方式访问广播变量携带的数据内容。累加器也是由Driver定义的，但Driver并不会向累加器中写入任何数据内容，累加器的内容更新，完全是由各个Executors以只写（Write only）的方式来完成，而Driver仅以只读的方式对更新后的内容进行访问。
|
||||
|
||||
关于广播变量，你首先需要掌握它的基本用法。给定任意类型的普通变量，你都可以使用SparkContext下面的broadcast API来创建广播变量。接下来，在RDD的转换与计算过程中，你可以通过调用广播变量的value函数，来访问封装的数据内容，从而辅助RDD的数据处理。
|
||||
|
||||
需要额外注意的是，在Driver与Executors之间，普通变量的分发与存储，是以Task为粒度的，因此，它所引入的网络与内存开销，会成为作业执行性能的一大隐患。在使用广播变量的情况下，数据内容的分发粒度变为以Executors为单位。相比前者，广播变量的优势高下立判，它可以大幅度消除前者引入的网络与内存开销，进而在整体上提升作业的执行效率。
|
||||
|
||||
关于累加器，首先你要清楚它的适用场景，当你需要做全局计数的时候，累加器会是个很好的帮手。其次，你需要掌握累加器的具体用法，可以分为这样3步：
|
||||
|
||||
|
||||
使用SparkContext下的[long | double | collection]Accumulator来定义累加器；
|
||||
在RDD的转换过程中，调用add函数更新累加器状态；
|
||||
在作业完成后，调用value函数，获取累加器的全局结果。
|
||||
|
||||
|
||||
每课一练
|
||||
|
||||
|
||||
在使用累加器对空字符串做全局计数的代码中，请你用普通变量去替换累加器，试一试，在不使用累加器的情况，能否得到预期的计算结果？
|
||||
累加器提供了Long、Double和Collection三种类型的支持，那么广播变量在类型支持上有限制吗？除了普通类型、集合类型之外，广播变量还支持其他类型吗？比如，Spark支持在RDD之上创建广播变量吗？
|
||||
|
||||
|
||||
欢迎你在留言区跟我交流互动，也推荐你把这一讲的内容分享给你身边的朋友，说不定就能帮他解决一个难题。
|
||||
|
||||
|
||||
|
||||
|
||||
@ -0,0 +1,235 @@
|
||||
|
||||
|
||||
因收到Google相关通知，网站将会择期关闭。相关通知内容
|
||||
|
||||
|
||||
20 Hive + Spark强强联合：分布式数仓的不二之选
|
||||
你好，我是吴磊。
|
||||
|
||||
在数据源与数据格式，以及数据转换那两讲（第15、16讲），我们介绍了在Spark SQL之上做数据分析应用开发的一般步骤。
|
||||
|
||||
这里我们简单回顾一下：首先，我们通过SparkSession read API从分布式文件系统创建DataFrame。然后，通过创建临时表并使用SQL语句，或是直接使用DataFrame API，来进行各式各样的数据转换、过滤、聚合等操作。最后，我们再用SparkSession的write API把计算结果写回分布式文件系统。
|
||||
|
||||
实际上，直接与文件系统交互，仅仅是Spark SQL数据应用的常见场景之一。Spark SQL另一类非常典型的场景是与Hive做集成、构建分布式数据仓库。我们知道，数据仓库指的是一类带有主题、聚合层次较高的数据集合，它的承载形式，往往是一系列Schema经过精心设计的数据表。在数据分析这类场景中，数据仓库的应用非常普遍。
|
||||
|
||||
在Hive与Spark这对“万金油”组合中，Hive擅长元数据管理，而Spark的专长是高效的分布式计算，二者的结合可谓是“强强联合”。今天这一讲，我们就来聊一聊Spark与Hive集成的两类方式，一类是从Spark的视角出发，我们称之为Spark with Hive；而另一类，则是从Hive的视角出发，业界的通俗说法是：Hive on Spark。
|
||||
|
||||
Hive架构与基本原理
|
||||
|
||||
磨刀不误砍柴工，在讲解这两类集成方式之前，我们不妨先花点时间，来了解一下Hive的架构和工作原理，避免不熟悉Hive的同学听得云里雾里。
|
||||
|
||||
Hive是Apache Hadoop社区用于构建数据仓库的核心组件，它负责提供种类丰富的用户接口，接收用户提交的SQL查询语句。这些查询语句经过Hive的解析与优化之后，往往会被转化为分布式任务，并交付Hadoop MapReduce付诸执行。
|
||||
|
||||
Hive是名副其实的“集大成者”，它的核心部件，其实主要是User Interface（1）和Driver（3）。而不论是元数据库（4）、存储系统（5），还是计算引擎（6），Hive都以“外包”、“可插拔”的方式交给第三方独立组件，所谓“把专业的事交给专业的人去做”，如下图所示。
|
||||
|
||||
|
||||
|
||||
Hive的User Interface为开发者提供SQL接入服务，具体的接入途径有Hive Server 2（2）、CLI和Web Interface（Web界面入口）。其中，CLI与Web Interface直接在本地接收SQL查询语句，而Hive Server 2则通过提供JDBC/ODBC客户端连接，允许开发者从远程提交SQL查询请求。显然，Hive Server 2的接入方式更为灵活，应用也更为广泛。
|
||||
|
||||
我们以响应一个SQL查询为例，看一看Hive是怎样工作的。接收到SQL查询之后，Hive的Driver首先使用其Parser组件，将查询语句转化为AST（Abstract Syntax Tree，查询语法树）。
|
||||
|
||||
紧接着，Planner组件根据AST生成执行计划，而Optimizer则进一步优化执行计划。要完成这一系列的动作，Hive必须要能拿到相关数据表的元信息才行，比如表名、列名、字段类型、数据文件存储路径、文件格式，等等。而这些重要的元信息，通通存储在一个叫作“Hive Metastore”（4）的数据库中。
|
||||
|
||||
本质上，Hive Metastore其实就是一个普通的关系型数据库（RDBMS），它可以是免费的MySQL、Derby，也可以是商业性质的Oracle、IBM DB2。实际上，除了用于辅助SQL语法解析、执行计划的生成与优化，Metastore的重要作用之一，是帮助底层计算引擎高效地定位并访问分布式文件系统中的数据源。
|
||||
|
||||
这里的分布式文件系统，可以是Hadoop生态的HDFS，也可以是云原生的Amazon S3。而在执行方面，Hive目前支持3类计算引擎，分别是Hadoop MapReduce、Tez和Spark。
|
||||
|
||||
当Hive采用Spark作为底层的计算引擎时，我们就把这种集成方式称作“Hive on Spark”。相反，当Spark仅仅是把Hive当成是一种元信息的管理工具时，我们把Spark与Hive的这种集成方式，叫作“Spark with Hive”。
|
||||
|
||||
你可能会觉得很困惑：“这两种说法听上去差不多嘛，两种集成方式，到底有什么本质的不同呢？”接下来，我们就按照“先易后难”的顺序，先来说说“Spark with Hive”这种集成方式，然后再去介绍“Hive on Spark”。
|
||||
|
||||
Spark with Hive
|
||||
|
||||
在开始正式学习Spark with Hive之前，我们先来说说这类集成方式的核心思想。前面我们刚刚说过，Hive Metastore利用RDBMS来存储数据表的元信息，如表名、表类型、表数据的Schema、表（分区）数据的存储路径、以及存储格式，等等。形象点说，Metastore就像是“户口簿”，它记录着分布式文件系统中每一份数据集的“底细”。
|
||||
|
||||
Spark SQL通过访问Hive Metastore这本“户口簿”，即可扩充数据访问来源。而这，就是Spark with Hive集成方式的核心思想。直白点说，在这种集成模式下，Spark是主体，Hive Metastore不过是Spark用来扩充数据来源的辅助工具。厘清Spark与Hive的关系，有助于我们后面区分Hive on Spark与Spark with Hive之间的差异。
|
||||
|
||||
作为开发者，我们可以通过3种途径来实现Spark with Hive的集成方式，它们分别是：
|
||||
|
||||
|
||||
创建SparkSession，访问本地或远程的Hive Metastore；
|
||||
通过Spark内置的spark-sql CLI，访问本地Hive Metastore；
|
||||
通过Beeline客户端，访问Spark Thrift Server。
|
||||
|
||||
|
||||
SparkSession + Hive Metastore
|
||||
|
||||
为了更好地理解Hive与Spark的关系，我们先从第一种途径，也就是通过SparkSession访问Hive Metastore说起。首先，我们使用如下命令来启动Hive Metastore。
|
||||
|
||||
hive --service metastore
|
||||
|
||||
|
||||
Hive Metastore启动之后，我们需要让Spark知道Metastore的访问地址，也就是告诉他数据源的“户口簿”藏在什么地方。
|
||||
|
||||
要传递这个消息，我们有两种办法。一种是在创建SparkSession的时候，通过config函数来明确指定hive.metastore.uris参数。另一种方法是让Spark读取Hive的配置文件hive-site.xml，该文件记录着与Hive相关的各种配置项，其中就包括hive.metastore.uris这一项。把hive-site.xml拷贝到Spark安装目录下的conf子目录，Spark即可自行读取其中的配置内容。
|
||||
|
||||
接下来，我们通过一个小例子，来演示第一种用法。假设Hive中有一张名为“salaries”的薪资表，每条数据都包含id和salary两个字段，表数据存储在HDFS，那么，在spark-shell中敲入下面的代码，我们即可轻松访问Hive中的数据表。
|
||||
|
||||
import org.apache.spark.sql.SparkSession
|
||||
import org.apache.spark.sql.DataFrame
|
||||
|
||||
val hiveHost: String = _
|
||||
// 创建SparkSession实例
|
||||
val spark = SparkSession.builder()
|
||||
.config("hive.metastore.uris", s"thrift://hiveHost:9083")
|
||||
.enableHiveSupport()
|
||||
.getOrCreate()
|
||||
|
||||
// 读取Hive表，创建DataFrame
|
||||
val df: DataFrame = spark.sql(“select * from salaries”)
|
||||
|
||||
df.show
|
||||
|
||||
/** 结果打印
|
||||
+---+------+
|
||||
| id|salary|
|
||||
+---+------+
|
||||
| 1| 26000|
|
||||
| 2| 30000|
|
||||
| 4| 25000|
|
||||
| 3| 20000|
|
||||
+---+------+
|
||||
*/
|
||||
|
||||
|
||||
在[第16讲]，我们讲过利用createTempView函数从数据文件创建临时表的方法，临时表创建好之后，我们就可以使用SparkSession的sql API来提交SQL查询语句。连接到Hive Metastore之后，咱们就可以绕过第一步，直接使用sql API去访问Hive中现有的表，是不是很方便？
|
||||
|
||||
更重要的是，createTempView函数创建的临时表，它的生命周期仅限于Spark作业内部，这意味着一旦作业执行完毕，临时表也就不复存在，没有办法被其他应用复用。Hive表则不同，它们的元信息已经持久化到Hive Metastore中，不同的作业、应用、甚至是计算引擎，如Spark、Presto、Impala，等等，都可以通过Hive Metastore来访问Hive表。
|
||||
|
||||
总结下来，在SparkSession + Hive Metastore这种集成方式中，Spark对于Hive的访问，仅仅涉及到Metastore这一环节，对于Hive架构中的其他组件，Spark并未触及。换句话说，在这种集成方式中，Spark仅仅是“白嫖”了Hive的Metastore，拿到数据集的元信息之后，Spark SQL自行加载数据、自行处理，如下图所示。
|
||||
|
||||
|
||||
|
||||
在第一种集成方式下，通过sql API，你可以直接提交复杂的SQL语句，也可以在创建DataFrame之后，再使用第16讲提到的各种算子去实现业务逻辑。
|
||||
|
||||
spark-sql CLI + Hive Metastore
|
||||
|
||||
不过，你可能会说：“既然是搭建数仓，那么能不能像使用普通数据库那样，直接输入SQL查询，绕过SparkSession的sql API呢？”
|
||||
|
||||
答案自然是肯定的，接下来，我们就来说说Spark with Hive的第二种集成方式：spark-sql CLI + Hive Metastore。与spark-shell、spark-submit类似，spark-sql也是Spark内置的系统命令。将配置好hive.metastore.uris参数的hive-site.xml文件放到Spark安装目录的conf下，我们即可在spark-sql中直接使用SQL语句来查询或是处理Hive表。
|
||||
|
||||
显然，在这种集成模式下，Spark和Hive的关系，与刚刚讲的SparkSession + Hive Metastore一样，本质上都是Spark通过Hive Metastore来扩充数据源。
|
||||
|
||||
不过，相比前者，spark-sql CLI的集成方式多了一层限制，那就是在部署上，spark-sql CLI与Hive Metastore必须安装在同一个计算节点。换句话说，spark-sql CLI只能在本地访问Hive Metastore，而没有办法通过远程的方式来做到这一点。
|
||||
|
||||
在绝大多数的工业级生产系统中，不同的大数据组件往往是单独部署的，Hive与Spark也不例外。由于Hive Metastore可用于服务不同的计算引擎，如前面提到的Presto、Impala，因此为了减轻节点的工作负载，Hive Metastore往往会部署到一台相对独立的计算节点。
|
||||
|
||||
在这样的背景下，不得不说，spark-sql CLI本地访问的限制，极大地削弱了它的适用场景，这也是spark-sql CLI + Hive Metastore这种集成方式几乎无人问津的根本原因。不过，这并不妨碍我们学习并了解它，这有助于我们对Spark与Hive之间的关系加深理解。
|
||||
|
||||
Beeline + Spark Thrift Server
|
||||
|
||||
说到这里，你可能会追问：“既然spark-sql CLI有这样那样的限制，那么，还有没有其他集成方式，既能够部署到生产系统，又能让开发者写SQL查询呢？”答案自然是“有”，Spark with Hive集成的第三种途径，就是使用Beeline客户端，去连接Spark Thrift Server，从而完成Hive表的访问与处理。
|
||||
|
||||
Beeline原本是Hive客户端，通过JDBC接入Hive Server 2。Hive Server 2可以同时服务多个客户端，从而提供多租户的Hive查询服务。由于Hive Server 2的实现采用了Thrift RPC协议框架，因此很多时候我们又把Hive Server 2称为“Hive Thrift Server 2”。
|
||||
|
||||
通过Hive Server 2接入的查询请求，经由Hive Driver的解析、规划与优化，交给Hive搭载的计算引擎付诸执行。相应地，查询结果再由Hiver Server 2返还给Beeline客户端，如下图右侧的虚线框所示。
|
||||
|
||||
|
||||
|
||||
Spark Thrift Server脱胎于Hive Server 2，在接收查询、多租户服务、权限管理等方面，这两个服务端的实现逻辑几乎一模一样。它们最大的不同，在于SQL查询接入之后的解析、规划、优化与执行。
|
||||
|
||||
我们刚刚说过，Hive Server 2的“后台”是Hive的那套基础架构。而SQL查询在接入到Spark Thrift Server之后，它首先会交由Spark SQL优化引擎进行一系列的优化。
|
||||
|
||||
在第14讲我们提过，借助于Catalyst与Tungsten这对“左膀右臂”，Spark SQL对SQL查询语句先后进行语法解析、语法树构建、逻辑优化、物理优化、数据结构优化、以及执行代码优化，等等。然后，Spark SQL将优化过后的执行计划，交付给Spark Core执行引擎付诸运行。
|
||||
|
||||
|
||||
|
||||
不难发现，SQL查询在接入Spark Thrift Server之后的执行路径，与DataFrame在Spark中的执行路径是完全一致的。
|
||||
|
||||
理清了Spark Thrift Server与Hive Server 2之间的区别与联系之后，接下来，我们来说说Spark Thrift Server的启动与Beeline的具体用法。要启动Spark Thrift Server，我们只需调用Spark提供的start-thriftserver.sh脚本即可。
|
||||
|
||||
// SPARK_HOME环境变量，指向Spark安装目录
|
||||
cd $SPARK_HOME/sbin
|
||||
|
||||
// 启动Spark Thrift Server
|
||||
./start-thriftserver.sh
|
||||
|
||||
|
||||
脚本执行成功之后，Spark Thrift Server默认在10000端口监听JDBC/ODBC的连接请求。有意思的是，关于监听端口的设置，Spark复用了Hive的hive.server2.thrift.port参数。与其他的Hive参数一样，hive.server2.thrift.port同样要在hive-site.xml配置文件中设置。
|
||||
|
||||
一旦Spark Thrift Server启动成功，我们就可以在任意节点上通过Beeline客户端来访问该服务。在客户端与服务端之间成功建立连接（Connections）之后，咱们就能在Beeline客户端使用SQL语句处理Hive表了。需要注意的是，在这种集成模式下，SQL语句背后的优化与计算引擎是Spark。
|
||||
|
||||
/**
|
||||
用Beeline客户端连接Spark Thrift Server，
|
||||
其中，hostname是Spark Thrift Server服务所在节点
|
||||
*/
|
||||
beeline -u “jdbc:hive2://hostname:10000”
|
||||
|
||||
|
||||
好啦，到此为止，Spark with Hive这类集成方式我们就讲完了。
|
||||
|
||||
为了巩固刚刚学过的内容，咱们趁热打铁，一起来做个简单的小结。不论是SparkSession + Hive Metastore、spark-sql CLI + Hive Metastore，还是Beeline + Spark Thrift Server，Spark扮演的角色都是执行引擎，而Hive的作用主要在于通过Metastore提供底层数据集的元数据。不难发现，在这类集成方式中，Spark唱“主角”，而Hive唱“配角”。
|
||||
|
||||
Hive on Spark
|
||||
|
||||
说到这里，你可能会好奇：“对于Hive社区与Spark社区来说，大家都是平等的，那么有没有Hive唱主角，而Spark唱配角的时候呢？”还真有，这就是Spark与Hive集成的另一种形式：Hive on Spark。
|
||||
|
||||
基本原理
|
||||
|
||||
在这一讲的开头，我们简单介绍了Hive的基础架构。Hive的松耦合设计，使得它的Metastore、底层文件系统、以及执行引擎都是可插拔、可替换的。
|
||||
|
||||
在执行引擎方面，Hive默认搭载的是Hadoop MapReduce，但它同时也支持Tez和Spark。所谓的“Hive on Spark”，实际上指的就是Hive采用Spark作为其后端的分布式执行引擎，如下图所示。
|
||||
|
||||
|
||||
|
||||
从用户的视角来看，使用Hive on MapReduce或是Hive on Tez与使用Hive on Spark没有任何区别，执行引擎的切换对用户来说是完全透明的。不论Hive选择哪一种执行引擎，引擎仅仅负责任务的分布式计算，SQL语句的解析、规划与优化，通通由Hive的Driver来完成。
|
||||
|
||||
为了搭载不同的执行引擎，Hive还需要做一些简单的适配，从而把优化过的执行计划“翻译”成底层计算引擎的语义。
|
||||
|
||||
举例来说，在Hive on Spark的集成方式中，Hive在将SQL语句转换为执行计划之后，还需要把执行计划“翻译”成RDD语义下的DAG，然后再把DAG交付给Spark Core付诸执行。从第14讲到现在，我们一直在强调，Spark SQL除了扮演数据分析子框架的角色之外，还是Spark新一代的优化引擎。
|
||||
|
||||
在Hive on Spark这种集成模式下，Hive与Spark衔接的部分是Spark Core，而不是Spark SQL，这一点需要我们特别注意。这也是为什么，相比Hive on Spark，Spark with Hive的集成在执行性能上会更胜一筹。毕竟，Spark SQL + Spark Core这种原装组合，相比Hive Driver + Spark Core这种适配组合，在契合度上要更高一些。
|
||||
|
||||
集成实现
|
||||
|
||||
分析完原理之后，接下来，我们再来说说，Hive on Spark的集成到底该怎么实现。
|
||||
|
||||
首先，既然我们想让Hive搭载Spark，那么我们事先得准备好一套完备的Spark部署。对于Spark的部署模式，Hive不做任何限定，Spark on Standalone、Spark on Yarn或是Spark on Kubernetes都是可以的。
|
||||
|
||||
Spark集群准备好之后，我们就可以通过修改hive-site.xml中相关的配置项，来轻松地完成Hive on Spark的集成，如下表所示。
|
||||
|
||||
|
||||
|
||||
其中，hive.execution.engine用于指定Hive后端执行引擎，可选值有“mapreduce”、“tez”和“spark”，显然，将该参数设置为“spark”，即表示采用Hive on Spark的集成方式。
|
||||
|
||||
确定了执行引擎之后，接下来我们自然要告诉Hive：“Spark集群部署在哪里”，spark.master正是为了实现这个目的。另外，为了方便Hive调用Spark的相关脚本与Jar包，我们还需要通过spark.home参数来指定Spark的安装目录。
|
||||
|
||||
配置好这3个参数之后，我们就可以用Hive SQL向Hive提交查询请求，而Hive则是先通过访问Metastore在Driver端完成执行计划的制定与优化，然后再将其“翻译”为RDD语义下的DAG，最后把DAG交给后端的Spark去执行分布式计算。
|
||||
|
||||
当你在终端看到“Hive on Spark”的字样时，就证明Hive后台的执行引擎确实是Spark，如下图所示。
|
||||
|
||||
|
||||
|
||||
当然，除了上述3个配置项以外，Hive还提供了更多的参数，用于微调它与Spark之间的交互。对于这些参数，你可以通过访问Hive on Spark配置项列表来查看。不仅如此，在第12讲，我们详细介绍了Spark自身的基础配置项，这些配置项都可以配置到hive-site.xml中，方便你更细粒度地控制Hive与Spark之间的集成。
|
||||
|
||||
重点回顾
|
||||
|
||||
好啦，到此为止，今天的内容就全部讲完啦！内容有点多，我们一起来做个总结。
|
||||
|
||||
今天这一讲，你需要了解Spark与Hive常见的两类集成方式，Spark with Hive和Hive on Spark。前者由Spark社区主导，以Spark为主、Hive为辅；后者则由Hive社区主导，以Hive为主、Spark为辅。两类集成方式各有千秋，适用场景各有不同。
|
||||
|
||||
在Spark with Hive这类集成方式中，Spark主要是利用Hive Metastore来扩充数据源，从而降低分布式文件的管理与维护成本，如路径管理、分区管理、Schema维护，等等。
|
||||
|
||||
对于Spark with Hive，我们至少有3种途径来实现Spark与Hive的集成，分别是SparkSession + Hive Metastore，spark-sql CLI + Hive Metastore和Beeline + Spark Thrift Server。对于这3种集成方式，我把整理了表格，供你随时查看。
|
||||
|
||||
|
||||
|
||||
与Spark with Hive相对，另一类集成方式是Hive on Spark。这种集成方式，本质上是Hive社区为Hive用户提供了一种新的选项，这个选项就是，在执行引擎方面，除了原有的MapReduce与Tez，开发者还可以选择执行性能更佳的Spark。
|
||||
|
||||
因此，在Spark大行其道的当下，习惯使用Hive的团队与开发者，更愿意去尝试和采用Spark作为后端的执行引擎。
|
||||
|
||||
熟悉了不同集成方式的区别与适用场景之后，在日后的工作中，当你需要将Spark与Hive做集成的时候，就可以做到有的放矢、有章可循，加油。
|
||||
|
||||
每课一练
|
||||
|
||||
|
||||
在Hive on Spark的部署模式下，用另外一套Spark部署去访问Hive Metastore，比如，通过创建SparkSession并访问Hive Metastore来扩充数据源。那么，在这种情况下，你能大概说一说用户代码的执行路径吗？
|
||||
|
||||
尽管咱们专栏的主题是Spark，但我强烈建议你学习并牢记Hive的架构设计。松耦合的设计理念让Hive本身非常轻量的同时，还给予了Hive极大的扩展能力。也正因如此，Hive才能一直牢牢占据开源数仓霸主的地位。Hive的设计思想是非常值得我们好好学习的，这样的设计思想可以推而广之，应用到任何需要考虑架构设计的地方，不论是前端、后端，还是大数据与机器学习。
|
||||
|
||||
|
||||
欢迎你在留言区跟我交流互动，也欢迎把这一讲的内容分享给更多同事、朋友。
|
||||
|
||||
|
||||
|
||||
|
||||
@ -0,0 +1,186 @@
|
||||
|
||||
|
||||
因收到Google相关通知，网站将会择期关闭。相关通知内容
|
||||
|
||||
|
||||
32 Window操作&Watermark：流处理引擎提供了哪些优秀机制？
|
||||
你好，我是吴磊。
|
||||
|
||||
在上一讲，我们从原理的角度出发，学习了Structured Streaming的计算模型与容错机制。深入理解这些基本原理，会帮我们开发流处理应用打下坚实的基础。
|
||||
|
||||
在“流动的Word Count”[那一讲]，我们演示了在Structured Streaming框架下，如何做流处理开发的一般流程。基于readStream API与writeStream API，我们可以像读写DataFrame那样，轻松地从Source获取数据流，并把处理过的数据写入Sink。
|
||||
|
||||
今天这一讲，咱们从功能的视角出发，继续来聊一聊Structured Streaming流处理引擎都为开发者都提供了哪些特性与能力，让你更灵活地设计并实现流处理应用。
|
||||
|
||||
Structured Streaming怎样坐享其成？
|
||||
|
||||
学习过计算模型之后，我们知道，不管是Batch mode的多个Micro-batch、多个作业的执行方式，还是Continuous mode下的一个Long running job，这些作业的执行计划，最终都会交付给Spark SQL与Spark Core付诸优化与执行。
|
||||
|
||||
|
||||
|
||||
而这，会带来两个方面的收益。一方面，凡是Spark SQL支持的开发能力，不论是丰富的DataFrame算子，还是灵活的SQL查询，Structured Streaming引擎都可以拿来即用。基于之前学过的内容，我们可以像处理普通的DataFrame那样，对基于流数据构建的DataFrame做各式各样的转换与聚合。
|
||||
|
||||
另一方面，既然开发入口同为DataFrame，那么流处理应用同样能够享有Spark SQL提供的“性能红利”。在Spark SQL学习模块，我们学习过Catalyst优化器与Tungsten，这两个组件会对用户代码做高度优化，从而提升应用的执行性能。
|
||||
|
||||
因此，就框架的功能来说，我们可以简单地概括为，Spark SQL所拥有的能力，Structured Streaming都有。不过，除了基本的数据处理能力以外，为了更好地支持流计算场景，Structured Streaming引擎还提供了一些专门针对流处理的计算能力，比如说Window操作、Watermark与延迟数据处理，等等。
|
||||
|
||||
Window操作
|
||||
|
||||
我们先来说说Window操作，它指的是，Structured Streaming引擎会基于一定的时间窗口，对数据流中的消息进行消费并处理。这是什么意思呢？首先，我们需要了解两个基本概念：Event Time和Processing Time，也即事件时间和处理时间。
|
||||
|
||||
所谓事件时间，它指的是消息生成的时间，比如，我们在netcat中敲入“Apache Spark”的时间戳是“2021-10-01 09:30:00”，那么这个时间，就是消息“Apache Spark”的事件时间。
|
||||
|
||||
|
||||
|
||||
而处理时间，它指的是，这个消息到达Structured Streaming引擎的时间，因此也有人把处理时间称作是到达时间（Arrival Time），也即消息到达流处理系统的时间。显然，处理时间要滞后于事件时间。
|
||||
|
||||
所谓Window操作，实际上就是Structured Streaming引擎基于事件时间或是处理时间，以固定间隔划定时间窗口，然后以窗口为粒度处理消息。在窗口的划分上，Structured Streaming支持两种划分方式，一种叫做Tumbling Window，另一种叫做Sliding Window。
|
||||
|
||||
我们可以用一句话来记住二者之间的区别，Tumbling Window划分出来的时间窗口“不重不漏”，而Sliding Window划分出来的窗口，可能会重叠、也可能会有遗漏，如下图所示。
|
||||
|
||||
|
||||
|
||||
不难发现，Sliding Window划分出来的窗口是否存在“重、漏”，取决于窗口间隔Interval与窗口大小Size之间的关系。Tumbling Window与Sliding Window并无优劣之分，完全取决于应用场景与业务需要。
|
||||
|
||||
干讲理论总是枯燥无趣，接下来，咱们对之前的“流动的Word Count”稍作调整，来演示Structured Streaming中的Window操作。为了让演示的过程更加清晰明了，这里我们采用Tumbling Window的划分方式，Sliding Window留给你作为课后作业。
|
||||
|
||||
为了完成实验，我们还是需要准备好两个终端。第一个终端用于启动spark-shell并提交流处理代码，而第二个终端用于启动netcat、输入数据流。要基于窗口去统计单词，我们仅需调整数据处理部分的代码，readStream与writeStream（Update Mode）部分的代码不需要任何改动。因此，为了聚焦Window操作的学习，我这里仅贴出了有所变动的部分。
|
||||
|
||||
df = df.withColumn("inputs", split($"value", ","))
|
||||
// 提取事件时间
|
||||
.withColumn("eventTime", element_at(col("inputs"),1).cast("timestamp"))
|
||||
// 提取单词序列
|
||||
.withColumn("words", split(element_at(col("inputs"),2), " "))
|
||||
// 拆分单词
|
||||
.withColumn("word", explode($"words"))
|
||||
// 按照Tumbling Window与单词做分组
|
||||
.groupBy(window(col("eventTime"), "5 minute"), col("word"))
|
||||
// 统计计数
|
||||
.count()
|
||||
|
||||
|
||||
为了模拟事件时间，我们在netcat终端输入的消息，会同时包含时间戳和单词序列。两者之间以逗号分隔，而单词与单词之间，还是用空格分隔，如下表所示。
|
||||
|
||||
|
||||
|
||||
因此，对于输入数据的处理，我们首先要分别提取出时间戳和单词序列，然后再把单词序列展开为单词。接下来，我们按照时间窗口与单词做分组，这里需要我们特别关注这行代码：
|
||||
|
||||
// 按照Tumbling Window与单词做分组
|
||||
.groupBy(window(col("eventTime"), "5 minute"), col("word"))
|
||||
|
||||
|
||||
其中window(col(“eventTime”), “5 minute”)的含义，就是以事件时间为准，以5分钟为间隔，创建Tumbling时间窗口。显然，window函数的第一个参数，就是创建窗口所依赖的时间轴，而第二个参数，则指定了窗口大小Size。说到这里，你可能会问：“如果我想创建Sliding Window，该怎么做呢？”
|
||||
|
||||
其实非常简单，只需要在window函数的调用中，再添加第三个参数即可，也就是窗口间隔Interval。比如说，我们还是想创建大小为5分钟的窗口，但是使用以3分钟为间隔进行滑动的方式去创建，那么我们就可以这样来实现：window(col(“eventTime”), “5 minute”, “3 minute”)。是不是很简单？
|
||||
|
||||
完成基于窗口和单词的分组之后，我们就可以继续调用count来完成计数了。不难发现，代码中的大多数转换操作，实际上都是我们常见的DataFrame算子，这也印证了这讲开头说的，Structured Streaming先天优势就是能坐享其成，享有Spark SQL提供的“性能红利”。
|
||||
|
||||
代码准备好之后，我们就可以把它们陆续敲入到spark-shell，并等待来自netcat的数据流。切换到netcat终端，并陆续（注意，是陆续！）输入刚刚的文本内容，我们就可以在spark-shell终端看到如下的计算结果。
|
||||
|
||||
|
||||
|
||||
可以看到，与“流动的Word Count”不同，这里的统计计数，是以窗口（5分钟）为粒度的。对于每一个时间窗口来说，Structured Streaming引擎都会把事件时间落入该窗口的单词统计在内。不难推断，随着时间向前推进，已经计算过的窗口，将不会再有状态上的更新。
|
||||
|
||||
比方说，当引擎处理到“2021-10-01 09:39:00,Spark Streaming”这条消息（记作消息39）时，理论上，前一个窗口“{2021-10-01 09:30:00, 2021-10-01 09:35:00}”（记作窗口30-35）的状态，也就是不同单词的统计计数，应该不会再有变化。
|
||||
|
||||
说到这里，你可能会有这样的疑问：“那不见得啊！如果在消息39之后，引擎又接收到一条事件时间落在窗口30-35的消息，那该怎么办呢？”要回答这个问题，我们还得从Late data和Structured Streaming的Watermark机制说起。
|
||||
|
||||
Late data与Watermark
|
||||
|
||||
我们先来说Late data，所谓Late data，它指的是那些事件时间与处理时间不一致的消息。虽然听上去有点绕，但通过下面的图解，我们就能瞬间理解Late data的含义。
|
||||
|
||||
|
||||
|
||||
通常来说，消息生成的时间，与消息到达流处理引擎的时间，应该是一致的。也即先生成的消息先到达，而后生成的消息后到达，就像上图中灰色部分消息所示意的那样。
|
||||
|
||||
不过，在现实情况中，总会有一些消息，因为网络延迟或者这样那样的一些原因，它们的处理时间与事件时间存在着比较大的偏差。这些消息到达引擎的时间，甚至晚于那些在它们之后才生成的消息。像这样的消息，我们统称为“Late data”，如图中红色部分的消息所示。
|
||||
|
||||
由于有Late data的存在，流处理引擎就需要一个机制，来判定Late data的有效性，从而决定是否让晚到的消息，参与到之前窗口的计算。
|
||||
|
||||
就拿红色的“Spark is cool”消息来说，在它到达Structured Streaming引擎的时候，属于它的事件时间窗口“{2021-10-01 09:30:00, 2021-10-01 09:35:00}”已经关闭了。那么，在这种情况下，Structured Streaming到底要不要用消息“Spark is cool”中的单词，去更新窗口30-35的状态（单词计数）呢？
|
||||
|
||||
为了解决Late data的问题，Structured Streaming采用了一种叫作Watermark的机制来应对。为了让你能够更容易地理解Watermark机制的原理，在去探讨它之前，我们先来澄清两个极其相似但是又完全不同的概念：水印和水位线。
|
||||
|
||||
要说清楚水印和水位线，咱们不妨来做个思想实验。假设桌子上有一盒鲜牛奶、一个吸管、还有一个玻璃杯。我们把盒子开个口，把牛奶全部倒入玻璃杯，接着，把吸管插入玻璃杯，然后通过吸管喝一口新鲜美味的牛奶。好啦，实验做完了，接下来，我们用它来帮我们澄清概念。
|
||||
|
||||
|
||||
|
||||
如图所示，最开始的时候，我们把牛奶倒到水印标示出来的高度，然后用吸管喝牛奶。不过，不论我们通过吸管喝多少牛奶，水印位置的牛奶痕迹都不会消失，也就是说，水印的位置是相对固定的。而水位线则不同，我们喝得越多，水位线下降得就越快，直到把牛奶喝光，水位线降低到玻璃杯底部。
|
||||
|
||||
好啦，澄清了水印与水位线的概念之后，我们还需要把这两个概念与流处理中的概念对应上。毕竟，“倒牛奶”的思想实验，是用来辅助我们学习Watermark机制的。
|
||||
|
||||
首先，水印与水位线，对标的都是消息的事件时间。水印相当于系统当前接收到的所有消息中最大的事件时间。而水位线指的是水印对应的事件时间，减去用户设置的容忍值。为了叙述方便，我们把这个容忍值记作T。在Structured Streaming中，我们把水位线对应的事件时间，称作Watermark，如下图所示。
|
||||
|
||||
|
||||
|
||||
显然，在流处理引擎不停地接收消息的过程中，水印与水位线也会相应地跟着变化。这个过程，跟我们刚刚操作的“倒牛奶、喝牛奶”的过程很像。每当新到消息的事件时间大于当前水印的时候，系统就会更新水印，这就好比我们往玻璃杯里倒牛奶，一直倒到最大事件时间的位置。然后，我们用吸管喝牛奶，吸掉深度为T的牛奶，让水位线下降到Watermark的位置。
|
||||
|
||||
把不同的概念关联上之后，接下来，我们来正式地介绍Structured Streaming的Watermark机制。我们刚刚说过，Watermark机制是用来决定，哪些Late data可以参与过往窗口状态的更新，而哪些Late data则惨遭抛弃。
|
||||
|
||||
如果用文字去解释Watermark机制，很容易把人说得云里雾里，因此，咱们不妨用一张流程图，来阐释这个过程。
|
||||
|
||||
|
||||
|
||||
可以看到，当有新消息到达系统后，Structured Streaming首先判断它的事件时间，是否大于水印。如果事件时间大于水印的话，Watermark机制则相应地更新水印与水位线，也就是最大事件时间与Watermark。
|
||||
|
||||
相反，假设新到消息的事件时间在当前水印以下，那么系统进一步判断消息的事件时间与“Watermark时间窗口下沿”的关系。所谓“Watermark时间窗口下沿”，它指的是Watermark所属时间窗口的起始时间。
|
||||
|
||||
咱们来举例说明，假设Watermark为“2021-10-01 09:34:00”，且事件时间窗口大小为5分钟，那么，Watermark所在时间窗口就是[“2021-10-01 09:30:00”，“2021-10-01 09:35:00”]，也即窗口30-35。这个时候，“Watermark时间窗口下沿”，就是窗口30-35的起始时间，也就是“2021-10-01 09:30:00”，如下图所示。
|
||||
|
||||
|
||||
|
||||
对于最新到达的消息，如果其事件时间大于“Watermark时间窗口下沿”，则消息可以参与过往窗口的状态更新，否则，消息将被系统抛弃，不再参与计算。换句话说，凡是事件时间小于“Watermark时间窗口下沿”的消息，系统都认为这样的消息来得太迟了，没有资格再去更新以往计算过的窗口。
|
||||
|
||||
不难发现，在这个过程中，延迟容忍度T是Watermark机制中的决定性因素，它决定了“多迟”的消息可以被系统容忍并接受。那么问题来了，既然T是由用户设定的，那么用户通过什么途径来设定这个T呢？再者，在Structured Streaming的开发框架下，Watermark机制要如何生效呢？
|
||||
|
||||
其实，要开启Watermark机制、并设置容忍度T，我们只需一行代码即可搞定。接下来，我们就以刚刚“带窗口的流动Word Count”为例，演示并说明Watermark机制的具体用法。
|
||||
|
||||
df = df.withColumn("inputs", split($"value", ","))
|
||||
// 提取事件时间
|
||||
.withColumn("eventTime", element_at(col("inputs"),1).cast("timestamp"))
|
||||
// 提取单词序列
|
||||
.withColumn("words", split(element_at(col("inputs"),2), " "))
|
||||
// 拆分单词
|
||||
.withColumn("word", explode($"words"))
|
||||
// 启用Watermark机制，指定容忍度T为10分钟
|
||||
.withWatermark("eventTime", "10 minute")
|
||||
// 按照Tumbling Window与单词做分组
|
||||
.groupBy(window(col("eventTime"), "5 minute"), col("word"))
|
||||
// 统计计数
|
||||
.count()
|
||||
|
||||
|
||||
可以看到，除了“.withWatermark(“eventTime”, “10 minute”)”这一句代码，其他部分与“带窗口的流动Word Count”都是一样的。这里我们用withWatermark函数来启用Watermark机制，该函数有两个参数，第一个参数是事件时间，而第二个参数就是由用户指定的容忍度T。
|
||||
|
||||
为了演示Watermark机制产生的效果，接下来，咱们对netcat输入的数据流做一些调整，如下表所示。注意，消息7“Test Test”和消息8“Spark is cool”都是Late data。
|
||||
|
||||
|
||||
|
||||
基于我们刚刚对于Watermark机制的分析，在容忍度T为10分钟的情况下，Late data消息8“Spark is cool”会被系统接受并消费，而消息7“Test Test”则将惨遭抛弃。你不妨先花点时间，自行推断出这一结论，然后再来看后面的结果演示。
|
||||
|
||||
|
||||
|
||||
上图中，左侧是输入消息7“Test Test”时spark-shell端的输出，可以看到，消息7被系统丢弃，没能参与计算。而右侧是消息8“Spark is cool”对应的执行结果，可以看到，“Spark”、“is”、“cool”这3个单词成功地更新了之前窗口30-35的状态（注意这里的“Spark”计数为3，而不是1）。
|
||||
|
||||
重点回顾
|
||||
|
||||
好啦，今天的内容，到这里就讲完了，我们一起来做个总结。首先，我们需要知道，在数据处理方面，Structured Streaming完全可以复用Spark SQL现有的功能与性能优势。因此，开发者完全可以“坐享其成”，使用DataFrame算子或是SQL语句，来完成流数据的处理。
|
||||
|
||||
再者，我们需要特别关注并掌握Structured Streaming的Window操作与Watermark机制。Structured Streaming支持两类窗口，一个是“不重不漏”的Tumbling Window，另一个是“可重可漏”的Sliding Window。二者并无高下之分，作为开发者，我们可以使用window函数，结合事件时间、窗口大小、窗口间隔等多个参数，来灵活地在两种窗口之间进行取舍。
|
||||
|
||||
对于Late data的处理，Structured Streaming使用Watermark机制来决定其是否参与过往窗口的计算与更新。关于Watermark机制的工作原理，我把它整理到了下面的流程图中，供你随时查看。
|
||||
|
||||
|
||||
|
||||
每课一练
|
||||
|
||||
|
||||
请你结合Tumbling Window的代码，把Tumbling Window改为Sliding Window。-
|
||||
对于Watermark机制中的示例，请你分析一下，为什么消息8“Spark is cool”会被系统接受并处理，而消息7“Test Test”却惨遭抛弃？
|
||||
|
||||
|
||||
欢迎你在留言区跟我交流讨论，也推荐你把这一讲分享给更多同事、朋友。
|
||||
|
||||
|
||||
|
||||
|
||||
@ -0,0 +1,421 @@
|
||||
|
||||
|
||||
因收到Google相关通知，网站将会择期关闭。相关通知内容
|
||||
|
||||
|
||||
34 Spark + Kafka：流计算中的“万金油”
|
||||
你好，我是吴磊。
|
||||
|
||||
在前面的几讲中，咱们不止一次提到，就数据源来说，Kafka是Structured Streaming最重要的Source之一。在工业级的生产系统中，Kafka与Spark这对组合最为常见。因此，掌握Kafka与Spark的集成，对于想从事流计算方向的同学来说，是至关重要的。
|
||||
|
||||
今天这一讲，咱们就来结合实例，说一说Spark与Kafka这对“万金油”组合如何使用。随着业务飞速发展，各家公司的集群规模都是有增无减。在集群规模暴涨的情况下，资源利用率逐渐成为大家越来越关注的焦点。毕竟，不管是自建的Data center，还是公有云，每台机器都是真金白银的投入。
|
||||
|
||||
实例：资源利用率实时计算
|
||||
|
||||
咱们今天的实例，就和资源利用率的实时计算有关。具体来说，我们首先需要搜集集群中每台机器的资源（CPU、内存）利用率，并将其写入Kafka。然后，我们使用Spark的Structured Streaming来消费Kafka数据流，并对资源利用率数据做初步的分析与聚合。最后，再通过Structured Streaming，将聚合结果打印到Console、并写回到Kafka，如下图所示。
|
||||
|
||||
|
||||
|
||||
一般来说，在工业级应用中，上图中的每一个圆角矩形，在部署上都是独立的。绿色矩形代表待监测的服务器集群，蓝色矩形表示独立部署的Kafka集群，而红色的Spark集群，也是独立部署的。所谓独立部署，它指的是，集群之间不共享机器资源，如下图所示。
|
||||
|
||||
|
||||
|
||||
如果你手头上没有这样的部署环境，也不用担心。要完成资源利用率实时计算的实例，咱们不必非要依赖独立部署的分布式集群。实际上，仅在单机环境中，你就可以复现今天的实例。
|
||||
|
||||
课程安排
|
||||
|
||||
今天这一讲涉及的内容比较多，在正式开始课程之前，咱们不妨先梳理一下课程内容，让你做到心中有数。
|
||||
|
||||
|
||||
|
||||
对于上图的1、2、3、4这四个步骤，我们会结合代码实现，分别讲解如下这四个环节：
|
||||
|
||||
|
||||
生成CPU与内存消耗数据流，写入Kafka；-
|
||||
Structured Streaming消费Kafka数据，并做初步聚合；-
|
||||
Structured Streaming将计算结果打印到终端；-
|
||||
Structured Streaming将计算结果写回Kafka，以备后用。
|
||||
|
||||
|
||||
除此之外，为了照顾不熟悉Kafka的同学，咱们还会对Kafka的安装、Topic创建与消费、以及Kafka的基本概念，做一个简单的梳理。
|
||||
|
||||
速读Kafka的架构与运行机制
|
||||
|
||||
在完成前面交代的计算环节之前，我们需要了解Kafka都提供了哪些核心功能。
|
||||
|
||||
在大数据的流计算生态中，Kafka是应用最为广泛的消息中间件（Messaging Queue）。消息中间件的核心功能有以下三点。
|
||||
|
||||
|
||||
连接消息生产者与消息消费者；-
|
||||
缓存生产者生产的消息（或者说事件）；-
|
||||
有能力让消费者以最低延迟访问到消息。
|
||||
|
||||
|
||||
所谓消息生产者，它指的是事件或消息的来源与渠道。在我们的例子中，待监测集群就是生产者。集群中的机器，源源不断地生产资源利用率消息。相应地，消息的消费者，它指的是访问并处理消息的系统。显然，在这一讲的例子中，消费者是Spark。Structured Streaming读取并处理Kafka中的资源利用率消息，对其进行聚合、汇总。
|
||||
|
||||
经过前面的分析，我们不难发现，消息中间件的存在，让生产者与消费者这两个系统之间，天然地享有如下三方面的收益。
|
||||
|
||||
|
||||
解耦：双方无需感知对方的存在，二者除了消息本身以外，再无交集；
|
||||
异步：双方都可以按照自己的“节奏”和“步调”，来生产或是消费消息，而不必受制于对方的处理能力；
|
||||
削峰：当消费者订阅了多个生产者的消息，且多个生产者同时生成大量消息时，得益于异步模式，消费者可以灵活地消费并处理消息，从而避免计算资源被撑爆的隐患。
|
||||
|
||||
|
||||
好啦，了解了Kafka的核心功能与特性之后，接下来，我们说一说Kafka的系统架构。与大多数主从架构的大数据组件（如HDFS、YARN、Spark、Presto、Flink，等等）不同，Kafka为无主架构。也就是说，在Kafka集群中，没有Master这样一个角色来维护全局的数据状态。
|
||||
|
||||
集群中的每台Server被称为Kafka Broker，Broker的职责在于存储生产者生产的消息，并为消费者提供数据访问。Broker与Broker之间，都是相互独立的，彼此不存在任何的依赖关系。
|
||||
|
||||
如果就这么平铺直叙去介绍Kafka架构的话，难免让你昏昏欲睡，所以我们上图解。配合示意图解释Kafka中的关键概念，会更加直观易懂。
|
||||
|
||||
|
||||
|
||||
刚刚说过，Kafka为无主架构，它依赖ZooKeeper来存储并维护全局元信息。所谓元信息，它指的是消息在Kafka集群中的分布与状态。在逻辑上，消息隶属于一个又一个的Topic，也就是消息的话题或是主题。在上面的示意图中，蓝色圆角矩形所代表的消息，全部隶属于Topic A；而绿色圆角矩形，则隶属于Topic B。
|
||||
|
||||
而在资源利用率的实例中，我们会创建两个Topic，一个是CPU利用率cpu-monitor，另一个是内存利用率mem-monitor。生产者在向Kafka写入消息的时候，需要明确指明，消息隶属于哪一个Topic。比方说，关于CPU的监控数据，应当发往cpu-monitor，而对于内存的监控数据，则应该发往mem-monitor。
|
||||
|
||||
为了平衡不同Broker之间的工作负载，在物理上，同一个Topic中的消息，以分区、也就是Partition为粒度进行存储，示意图中的圆角矩形，代表的正是一个个数据分区。在Kafka中，一个分区，实际上就是磁盘上的一个文件目录。而消息，则依序存储在分区目录的文件中。
|
||||
|
||||
为了提供数据访问的高可用（HA，High Availability），在生产者把消息写入主分区（Leader）之后，Kafka会把消息同步到多个分区副本（Follower），示意图中的步骤1与步骤2演示了这个过程。
|
||||
|
||||
一般来说，消费者默认会从主分区拉取并消费数据，如图中的步骤3所示。而当主分区出现故障、导致数据不可用时，Kafka就会从剩余的分区副本中，选拔出一个新的主分区来对外提供服务，这个过程，又称作“选主”。
|
||||
|
||||
好啦，到此为止，Kafka的基础功能和运行机制我们就讲完了，尽管这些介绍不足以覆盖Kafka的全貌，但是，对于初学者来说，这些概念足以帮我们进军实战，做好Kafka与Spark的集成。
|
||||
|
||||
Kafka与Spark集成
|
||||
|
||||
接下来，咱们就来围绕着“资源利用率实时计算”这个例子，手把手地带你实现Kafka与Spark的集成过程。首先，第一步，我们先来准备Kafka环境。
|
||||
|
||||
Kafka环境准备
|
||||
|
||||
要配置Kafka环境，我们只需要简单的三个步骤即可：
|
||||
|
||||
|
||||
安装ZooKeeper、安装Kafka，启动ZooKeeper；-
|
||||
修改Kafka配置文件server.properties，设置ZooKeeper相关配置项；-
|
||||
启动Kafka，创建Topic。
|
||||
|
||||
|
||||
首先，咱们从 ZooKeeper官网与 Kafka官网，分别下载二者的安装包。然后，依次解压安装包、并配置相关环境变量即可，如下表所示。
|
||||
|
||||
// 下载ZooKeeper安装包
|
||||
wget https://archive.apache.org/dist/zookeeper/zookeeper-3.7.0/apache-zookeeper-3.7.0-bin.tar.gz
|
||||
// 下载Kafka安装包
|
||||
wget https://archive.apache.org/dist/kafka/2.8.0/kafka_2.12-2.8.0.tgz
|
||||
|
||||
// 把ZooKeeper解压并安装到指定目录
|
||||
tar -zxvf apache-zookeeper-3.7.0-bin.tar.gz -C /opt/zookeeper
|
||||
// 把Kafka解压并安装到指定目录
|
||||
tar -zxvf kafka_2.12-2.8.0.tgz -C /opt/kafka
|
||||
|
||||
// 编辑环境变量
|
||||
vi ~/.bash_profile
|
||||
/** 输入如下内容到文件中
|
||||
export ZOOKEEPER_HOME=/opt/zookeeper/apache-zookeeper-3.7.0-bin
|
||||
export KAFKA_HOME=/opt/kafka/kafka_2.12-2.8.0
|
||||
export PATH=$PATH:$ZOOKEEPER_HOME/bin:$KAFKA_HOME/bin
|
||||
*/
|
||||
|
||||
// 启动ZooKeeper
|
||||
zkServer.sh start
|
||||
|
||||
|
||||
接下来，我们打开Kafka配置目录下（也即$KAFKA_HOME/config）的server.properties文件，将其中的配置项zookeeper.connect，设置为“hostname:2181”，也就是主机名加端口号。
|
||||
|
||||
如果你把ZooKeeper和Kafka安装到同一个节点，那么hostname可以写localhost。而如果是分布式部署，hostname要写ZooKeeper所在的安装节点。一般来说，ZooKeeper默认使用2181端口来提供服务，这里我们使用默认端口即可。
|
||||
|
||||
配置文件设置完毕之后，我们就可以使用如下命令，在多个节点启动Kafka Broker。
|
||||
|
||||
kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties
|
||||
|
||||
|
||||
Kafka启动之后，咱们就来创建刚刚提到的两个Topic：cpu-monitor和mem-monitor，它们分别用来存储CPU利用率消息与内存利用率消息。
|
||||
|
||||
kafka-topics.sh --zookeeper hostname:2181/kafka --create
|
||||
--topic cpu-monitor
|
||||
--replication-factor 3
|
||||
--partitions 1
|
||||
|
||||
kafka-topics.sh --zookeeper hostname:2181/kafka --create
|
||||
--topic mem-monitor
|
||||
--replication-factor 3
|
||||
--partitions 1
|
||||
|
||||
|
||||
怎么样？是不是很简单？要创建Topic，只要指定ZooKeeper服务地址、Topic名字和副本数量即可。不过，这里需要特别注意的是，副本数量，也就是replication-factor，不能超过集群中的Broker数量。所以，如果你是本地部署的话，也就是所有服务都部署到一台节点，那么这里的replication-factor应该设置为1。
|
||||
|
||||
好啦，到此为止，Kafka环境安装、配置完毕。下一步，我们就该让生产者去生产资源利用率消息，并把消息源源不断地注入Kafka集群了。
|
||||
|
||||
消息的生产
|
||||
|
||||
在咱们的实例中，我们要做的是监测集群中每台机器的资源利用率。因此，我们需要这些机器，每隔一段时间，就把CPU和内存利用率发送出来。而要做到这一点，咱们只需要完成一下两个两个必要步骤：
|
||||
|
||||
|
||||
每台节点从本机收集CPU与内存使用数据；-
|
||||
|
||||
把收集到的数据，按照固定间隔，发送给Kafka集群。-
|
||||
由于消息生产这部分代码比较长，而我们的重点是学习Kafka与Spark的集成，因此，这里咱们只给出这两个步骤所涉及的关键代码片段。完整的代码实现，你可以从这里进行下载。
|
||||
|
||||
import java.lang.management.ManagementFactory
|
||||
import java.lang.reflect.Modifier
|
||||
|
||||
def getUsage(mothedName: String): Any = {
|
||||
// 获取操作系统Java Bean
|
||||
val operatingSystemMXBean = ManagementFactory.getOperatingSystemMXBean
|
||||
// 获取操作系统对象中声明过的方法
|
||||
|
||||
for (method <- operatingSystemMXBean.getClass.getDeclaredMethods) {
|
||||
method.setAccessible(true)
|
||||
|
||||
// 判断是否为我们需要的方法名
|
||||
|
||||
if (method.getName.startsWith(mothedName) && Modifier.isPublic(method.getModifiers)) {
|
||||
|
||||
// 调用并执行方法，获取指定资源（CPU或内存）的利用率
|
||||
|
||||
return method.invoke(operatingSystemMXBean)
|
||||
}
|
||||
}
|
||||
throw new Exception(s"Can not reflect method: ${mothedName}")
|
||||
|
||||
}
|
||||
|
||||
// 获取CPU利用率
|
||||
def getCPUUsage(): String = {
|
||||
|
||||
var usage = 0.0
|
||||
|
||||
try{
|
||||
// 调用getUsage方法，传入”getSystemCpuLoad”参数，获取CPU利用率
|
||||
|
||||
usage = getUsage("getSystemCpuLoad").asInstanceOf[Double] * 100
|
||||
} catch {
|
||||
case e: Exception => throw e
|
||||
}
|
||||
usage.toString
|
||||
|
||||
}
|
||||
|
||||
// 获取内存利用率
|
||||
def getMemoryUsage(): String = {
|
||||
|
||||
var freeMemory = 0L
|
||||
var totalMemory = 0L
|
||||
var usage = 0.0
|
||||
|
||||
try{
|
||||
// 调用getUsage方法，传入相关内存参数，获取内存利用率
|
||||
|
||||
freeMemory = getUsage("getFreePhysicalMemorySize").asInstanceOf[Long]
|
||||
totalMemory = getUsage("getTotalPhysicalMemorySize").asInstanceOf[Long]
|
||||
|
||||
// 用总内存，减去空闲内存，获取当前内存用量
|
||||
|
||||
usage = (totalMemory - freeMemory.doubleValue) / totalMemory * 100
|
||||
} catch {
|
||||
case e: Exception => throw e
|
||||
}
|
||||
usage.toString
|
||||
|
||||
}
|
||||
|
||||
|
||||
利用Java的反射机制，获取资源利用率
|
||||
|
||||
上面的代码，用来获取CPU与内存利用率。在这段代码中，最核心的部分是利用Java的反射机制，来获取操作系统对象的各个公有方法，然后通过调用这些公有方法，来完成资源利用率的获取。
|
||||
|
||||
不过，看到这你可能会说：“我并不了解Java的反射机制，上面的代码看不太懂。”这也没关系，只要你能结合注释，把上述代码的计算逻辑搞清楚即可。获取到资源利用率的数据之后，接下来，我们就可以把它们发送给Kafka了。
|
||||
|
||||
import org.apache.kafka.clients.producer.{Callback, KafkaProducer, ProducerConfig, ProducerRecord}
|
||||
import org.apache.kafka.common.serialization.StringSerializer
|
||||
|
||||
// 初始化属性信息
|
||||
def initConfig(clientID: String): Properties = {
|
||||
val props = new Properties
|
||||
val brokerList = "localhost:9092"
|
||||
// 指定Kafka集群Broker列表
|
||||
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList)
|
||||
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer].getName)
|
||||
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer].getName)
|
||||
props.put(ProducerConfig.CLIENT_ID_CONFIG, clientID)
|
||||
props
|
||||
}
|
||||
|
||||
val clientID = "usage.monitor.client"
|
||||
val cpuTopic = "cpu-monitor"
|
||||
val memTopic = "mem-monitor"
|
||||
|
||||
// 定义属性，其中包括Kafka集群信息、序列化方法，等等
|
||||
val props = initConfig(clientID)
|
||||
// 定义Kafka Producer对象，用于发送消息
|
||||
val producer = new KafkaProducer[String, String](props)
|
||||
// 回调函数，可暂时忽略
|
||||
val usageCallback = _
|
||||
|
||||
while (true) {
|
||||
var cpuUsage = new String
|
||||
var memoryUsage = new String
|
||||
// 调用之前定义的函数，获取CPU、内存利用率
|
||||
cpuUsage = getCPUUsage()
|
||||
memoryUsage = getMemoryUsage()
|
||||
|
||||
// 为CPU Topic生成Kafka消息
|
||||
val cpuRecord = new ProducerRecord[String, String](cpuTopic, clientID, cpuUsage)
|
||||
// 为Memory Topic生成Kafka消息
|
||||
val memRecord = new ProducerRecord[String, String](memTopic, clientID, memoryUsage)
|
||||
// 向Kafka集群发送CPU利用率消息
|
||||
producer.send(cpuRecord, usageCallback)
|
||||
// 向Kafka集群发送内存利用率消息
|
||||
producer.send(memRecord, usageCallback)
|
||||
// 设置发送间隔：2秒
|
||||
Thread.sleep(2000)
|
||||
}
|
||||
|
||||
|
||||
从上面的代码中，我们不难发现，其中的关键步骤有三步：
|
||||
|
||||
|
||||
定义Kafka Producer对象，其中需要我们在属性信息中，指明Kafka集群相关信息；
|
||||
调用之前定义的函数getCPUUsage、getMemoryUsage，获取CPU与内存资源利用率；
|
||||
把资源利用率封装为消息，并发送给对应的Topic。-
|
||||
好啦，到此为止，生产端的事情，我们就全部做完啦。在待监测的集群中，每隔两秒钟，每台机器都会向Kafka集群的cpu-monitor和mem-monitor这两个Topic发送即时消息。Kafka接收到这些消息之后，会把它们落盘到相应的分区中，等待着下游（也就是Spark）的消费。
|
||||
|
||||
|
||||
消息的消费
|
||||
|
||||
接下来，终于要轮到Structured Streaming闪亮登场了。在流计算模块的[第一讲]，我们就提到，Structured Streaming支持多种Source（Socket、File、Kafka），而在这些Source中，Kafka的应用最为广泛。在用法上，相比其他Source，从Kafka接收并消费数据并没有什么两样，咱们依然是依赖“万能”的readStream API，如下表所示。
|
||||
|
||||
import org.apache.spark.sql.DataFrame
|
||||
|
||||
// 依然是依赖readStream API
|
||||
val dfCPU:DataFrame = spark.readStream
|
||||
// format要明确指定Kafka
|
||||
.format("kafka")
|
||||
// 指定Kafka集群Broker地址，多个Broker用逗号隔开
|
||||
.option("kafka.bootstrap.servers", "hostname1:9092,hostname2:9092,hostname3:9092")
|
||||
// 订阅相关的Topic，这里以cpu-monitor为例
|
||||
.option("subscribe", "cpu-monitor")
|
||||
.load()
|
||||
|
||||
|
||||
对于readStream API的用法，想必你早已烂熟于心了，上面的代码，你应该会觉得看上去很眼熟。这里需要我们特别注意的，主要有三点：
|
||||
|
||||
|
||||
format中需要明确指定Kafka；
|
||||
为kafka.bootstrap.servers键值指定Kafka集群Broker，多个Broker之间以逗号分隔；
|
||||
为subscribe键值指定需要消费的Topic名，明确Structured Streaming要消费的Topic。
|
||||
|
||||
|
||||
挥完上面的“三板斧”之后，我们就得到了用于承载CPU利用率消息的DataFrame。有了DataFrame，我们就可以利用Spark SQL提供的能力，去做各式各样的数据处理。再者，结合Structured Streaming框架特有的Window和Watermark机制，我们还能以时间窗口为粒度做计数统计，同时决定“多迟”的消息，我们将不再处理。
|
||||
|
||||
不过，在此之前，咱们不妨先来直观看下代码，感受一下存在Kafka中的消息长什么样子。
|
||||
|
||||
import org.apache.spark.sql.streaming.{OutputMode, Trigger}
|
||||
import scala.concurrent.duration._
|
||||
|
||||
dfCPU.writeStream
|
||||
.outputMode("Complete")
|
||||
// 以Console为Sink
|
||||
.format("console")
|
||||
// 每10秒钟，触发一次Micro-batch
|
||||
.trigger(Trigger.ProcessingTime(10.seconds))
|
||||
.start()
|
||||
.awaitTermination()
|
||||
|
||||
|
||||
利用上述代码，通过终端，我们可以直接观察到Structured Streaming获取的Kafka消息，从而对亟待处理的消息，建立一个感性的认知，如下图所示。
|
||||
|
||||
|
||||
|
||||
在上面的数据中，除了Key、Value以外，其他信息都是消息的元信息，也即消息所属Topic、所在分区、消息的偏移地址、录入Kafka的时间，等等。
|
||||
|
||||
在咱们的实例中，Key对应的是发送资源利用率数据的服务器节点，而Value则是具体的CPU或是内存利用率。初步熟悉了消息的Schema与构成之后，接下来，咱们就可以有的放矢地去处理这些实时的数据流了。
|
||||
|
||||
对于这些每两秒钟就产生的资源利用率数据，假设我们仅关心它们在一定时间内（比如10秒钟）的平均值，那么，我们就可以结合Trigger与聚合计算来做到这一点，代码如下所示。
|
||||
|
||||
import org.apache.spark.sql.types.StringType
|
||||
|
||||
dfCPU
|
||||
.withColumn("clientName", $"key".cast(StringType))
|
||||
.withColumn("cpuUsage", $"value".cast(StringType))
|
||||
// 按照服务器做分组
|
||||
.groupBy($"clientName")
|
||||
// 求取均值
|
||||
.agg(avg($"cpuUsage").cast(StringType).alias("avgCPUUsage"))
|
||||
.writeStream
|
||||
.outputMode("Complete")
|
||||
// 以Console为Sink
|
||||
.format("console")
|
||||
// 每10秒触发一次Micro-batch
|
||||
.trigger(Trigger.ProcessingTime(10.seconds))
|
||||
.start()
|
||||
.awaitTermination()
|
||||
|
||||
|
||||
可以看到，我们利用Fixed interval trigger，每隔10秒创建一个Micro-batch。然后，在一个Micro-batch中，我们按照发送消息的服务器做分组，并计算CPU利用率平均值。最后将统计结果打印到终端，如下图所示。
|
||||
|
||||
|
||||
|
||||
再次写入Kafka
|
||||
|
||||
实际上，除了把结果打印到终端外，我们还可以把它写回Kafka。我们知道Structured Streaming支持种类丰富的Sink，除了常用于测试的Console以外，还支持File、Kafka、Foreach(Batch)，等等。要把数据写回Kafka也不难，我们只需在writeStream API中，指定format为Kafka并设置相关选项即可，如下表所示。
|
||||
|
||||
dfCPU
|
||||
.withColumn("key", $"key".cast(StringType))
|
||||
.withColumn("value", $"value".cast(StringType))
|
||||
.groupBy($"key")
|
||||
.agg(avg($"value").cast(StringType).alias("value"))
|
||||
.writeStream
|
||||
.outputMode("Complete")
|
||||
// 指定Sink为Kafka
|
||||
.format("kafka")
|
||||
// 设置Kafka集群信息，本例中只有localhost一个Kafka Broker
|
||||
.option("kafka.bootstrap.servers", "localhost:9092")
|
||||
// 指定待写入的Kafka Topic，需事先创建好Topic：cpu-monitor-agg-result
|
||||
.option("topic", "cpu-monitor-agg-result")
|
||||
// 指定WAL Checkpoint目录地址
|
||||
.option("checkpointLocation", "/tmp/checkpoint")
|
||||
.trigger(Trigger.ProcessingTime(10.seconds))
|
||||
.start()
|
||||
.awaitTermination()
|
||||
|
||||
|
||||
我们首先指定Sink为Kafka，然后通过option选项，分别设置Kafka集群信息、待写入的Topic名字，以及WAL Checkpoint目录。将上述代码敲入spark-shell，Structured Streaming会每隔10秒钟，就从Kafka拉取原始的利用率信息（Topic：cpu-monitor），然后按照服务器做分组聚合，最终再把聚合结果写回到Kafka（Topic：cpu-monitor-agg-result）。
|
||||
|
||||
这里有两点需要特别注意，一个是读取与写入的Topic要分开，以免造成逻辑与数据上的混乱。再者，细心的你可能已经发现，写回Kafka的数据，在Schema上必须用“key”和“value”这两个固定的字段，而不能再像写入Console时，可以灵活地定义类似于“clientName”和“avgCPUUsage”这样的字段名，关于这一点，还需要你特别关注。
|
||||
|
||||
重点回顾
|
||||
|
||||
好啦，到此为止，我手把手地带你实现了Kafka与Spark的集成，完成了图中涉及的每一个环节，也即从消息的生产、到写入Kafka，再到消息的消费与处理，并最终写回Kafka。
|
||||
|
||||
|
||||
|
||||
今天的内容比较多，你除了需要掌握集成中的每一个环节与用法外，还需要了解一些有关Kafka的基本概念与特性。Kafka是应用最为广泛的消息中间件（Messaging Queue），它的核心功能有三个：
|
||||
|
||||
|
||||
连接消息生产者与消息消费者；-
|
||||
缓存生产者生产的消息（或者说事件）；-
|
||||
有能力让消费者以最低延迟访问到消息。-
|
||||
对于Kafka的一些基本概念，你无需死记硬背，在需要的时候，回顾后面这张架构图即可。这张图中，清楚地标记了Kafka的基础概念，以及消息生产、缓存与消费的简易流程。
|
||||
|
||||
|
||||
|
||||
|
||||
而对于Kafka与Spark两者的集成，不管是Structured Streaming通过readStream API消费Kafka消息，还是使用writeStream API将计算结果写入Kafka，你只需要记住如下几点，即可轻松地搭建这对“万金油”组合。
|
||||
|
||||
|
||||
在format函数中，指定Kafka为Source或Sink；
|
||||
在option选项中，为kafka.bootstrap.servers键值指定Kafka集群Broker；
|
||||
在option选项中，设置subscribe或是topic，指定读取或是写入的Kafka Topic。
|
||||
|
||||
|
||||
每课一练
|
||||
|
||||
请你结合本讲中CPU利用率的代码，针对内存利用率，完成示意图中的各个环节，也即内存利用率消息的生产、写入Kafka（步骤1）、消息的消费与计算（步骤2、3），聚合结果再次写入Kafka（步骤4）。
|
||||
|
||||
|
||||
|
||||
欢迎你把今天这讲内容转发给更多同事、朋友，跟他一起动手试验一下Spark + Kafka的实例，我再留言区等你分享。
|
||||
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user