发帖
雷达卡
面向初学者的Apache Spark流教程
介绍
在当今世界，我们以极快的速度生成数据的情况下，对数据进行正确的分析并在正确的时间提供有用和有意义的结果可以为处理数据产品的许多领域提供有用的解决方案。我们可以在医疗保健和金融领域将此应用到媒体，零售，旅行服务等。一些可靠的例子包括Netflix实时提供个性化推荐，亚马逊跟踪您与平台上不同产品的互动并立即提供相关产品，或者需要实时流式传输大量数据并对其进行不同分析的业务。
Apache Spark是一种可以实时处理大数据并执行不同分析的出色框架之一。在此博客中，我们将使用火花流技术大规模处理高速数据。我们将使用Kafka将数据提取到我们的Spark代码中
什么是星火？
Apache Spark是一种快如闪电的集群计算技术，专为快速计算而设计。它基于Hadoop MapReduce，并扩展了MapReduce模型以有效地将其用于更多类型的计算，其中包括交互式查询和流处理。Spark的主要功能是其内存中的群集计算，可提高应用程序的处理速度。
Spark旨在涵盖各种工作负载，例如批处理应用程序，迭代算法，交互式查询和流。除了在各自的系统中支持所有这些工作负载之外，它还减轻了维护单独工具的管理负担。
什么是Spark Streaming？
Spark Streaming是核心Spark API的扩展，可实现实时数据流的高吞吐量，容错流处理。数据可以从许多来源（例如Kafka，Flume，Twitter，ZeroMQ或TCP套接字）中提取，并可以使用复杂的算法进行处理，这些算法以高级功能（如map，reduce，join和window）表示。最后，可以将处理后的数据推送到文件系统，数据库和实时仪表板。由于Spark Streaming构建在Spark之上，因此用户可以在数据流上应用Spark的内置机器学习算法（MLlib）和图形处理算法（GraphX）。与其他流项目相比，Spark流具有以下功能和优点：
易用性：Spark Streaming将Spark的语言集成API带到流处理中，使用户可以使用Java，Python和Scala以与批处理作业相同的方式编写流应用程序。
容错：Spark Streaming能够检测由于节点或进程故障而导致的中途数据丢失并从中恢复。
Spark Streaming如何工作？
Spark Streaming通过将 流变成称为离散流 或 DStream的微批 。DStream是Spark Streaming提供的API，用于创建和 处理微批次。DStream只不过是像其他RDD一样在Spark的核心执行引擎上处理的一系列RDD。可以从任何流媒体源（例如Flume或Kafka）创建它。
Spark流和Spark结构化流之间的区别
Spark流 基于DStream。DStream由一系列连续的RDD表示，这是Spark对不变的分布式数据集的抽象。Spark Streaming具有以下问题。
困难-构建支持交付策略的流传输管道并非易事：只需保证一次，处理数据迟到或容错。当然，它们都是可以实现的，但是它们需要程序员的一些额外工作。
不一致-用于生成批处理（RDD，数据集）的API与流处理（DStream）的API不同。当然，没有什么阻碍代码编写的，但是至少处理抽象总是比较简单的（尤其是维护成本）。
Spark结构化流 可以理解为无限制的表，随着新的传入数据而增长，即可以认为是基于Spark SQL构建的流处理。
更具体地说，结构化流式传输为Spark带来了一些新概念。
一次保证-结构化流将重点放在该概念上。这意味着数据仅处理一次，并且输出不包含重复项。
事件时间-DStream流处理中观察到的问题之一是处理顺序，即较早生成的数据在较晚生成的数据之后进行处理的情况。结构化流使用称为事件时间的概念来处理此问题，在某些情况下，事件时间允许在处理管道中正确地聚合后期数据。
接收器，结果表，输出模式和水印是spark结构化流的其他功能。
实施目标
在此博客中，我们将尝试查找句子中存在的字数。这里的重点是这次这句话不会出现在文本文件中。句子将作为流数据点通过实时流传递。我们将对流动数据中出现的单词进行计数。在这种情况下，数据不是固定的而是在不断移动。它也被称为高速数据。在这种情况下，我们将即时计算字数！我们将使用Kafka将数据作为实时流传输。Spark具有可用于连接数据流（如Kafka）的不同连接器
使用Kafka的字数统计示例
为了从通过Kafka流入的数据中查找字数，我们需要执行几个步骤。
Spark和Kafka Connector的初始化
我们的主要任务是为我们的应用程序创建一个入口点。我们还需要在环境中设置和初始化Spark Streaming。这是通过以下代码完成的
1
2
val sparkConf = new SparkConf().setAppName("DirectKafkaWordCount")
val ssc = new StreamingContext(sparkConf
由于我们已经初始化了Spark Streaming，因此我们需要将我们的应用程序与Kafka连接以接收流动数据。Spark具有内置的连接器，可用于将您的应用程序与不同的消息队列连接。我们需要将信息（如主题名称）放在此处，以便从中使用数据。我们需要定义Kafka主题所在的引导服务器。提供所有必需的信息后，我们将使用createDirectStream函数建立与Kafka的连接。您可以在下面找到实现
1
2
3
4
5
6
7
8
9
10
val topicsSet = topics.split("
val wordCounts = words.map(x => (x
wordCounts.print()
最后，除非您使用Spark Streaming实例调用start函数，否则处理将不会开始。另外，请记住，您需要等待shutdown命令并保持代码运行以通过实时流接收数据。为此，我们使用了awaitTermination方法。您可以通过以下两行实现上述逻辑
1
2
ssc.start()
ssc.awaitTermination()
完整代码
package org.apache.spark.examples.streaming
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka010._
object DirectKafkaWordCount {
def main(args: Array[String]) {
if (args.length < 3) {
System.err.println(s"""
|Usage: DirectKafkaWordCount <brokers> <topics>
| <brokers> is a list of one or more Kafka brokers
| <groupId> is a consumer group name to consume from topics
| <topics> is a list of one or more kafka topics to consume from
|
".stripMargin)
System.exit(1)
}
StreamingExamples.setStreamingLogLevels()
val Array(brokers, groupId, topics) = args
// Create context with 2 second batch interval
val sparkConf = new SparkConf().setAppName(DirectKafkaWordCount")
val ssc = new StreamingContext(sparkConf
// Create direct kafka stream with brokers and topics
val topicsSet = topics.split("
val wordCounts = words.map(x => (x
wordCounts.print()
// Start the computation
ssc.start()
ssc.awaitTermination()
}
}
// scalastyle:on println
概要
早些时候，由于Hadoop具有高延迟，因此不适用于近实时处理 需求。在大多数情况下，我们使用Hadoop进行 批处理， 而使用Storm进行 流处理。它导致代码大小增加，许多错误修复，开发工作以及引起其他问题，这使大数据Hadoop与Apache Spark有所不同。
最终，Spark Streaming解决了所有这些问题。它提供了可扩展，高效，有弹性的集成系统。该模型为批处理和流提供了执行和统一编程。尽管迅速采用它的一个主要原因是统一的独特数据处理能力。它成为一个热蛋糕供开发者使用一个单一的框架来 实现 所有的加工需求。 另外，通过Spark SQL流数据可以与静态数据源结合。
1
提升卡
置顶卡
沉默卡
变色卡
抢沙发
千斤顶
显身卡
京公网安备 11010802022788号
