Spark Streaming（二十六）DStream基本数据源、高级数据源

2024-02-27 17:08
文章标签 基本 高级 二十六 数据源 spark streaming dstream

本文主要是介绍Spark Streaming（二十六）DStream基本数据源、高级数据源，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

输入DStream和Receiver

输入DStream其实就是从数据源接收到的输入数据流的DStream。每个DStream都与一个Receiver对象一一对应。SparkStreaming提供了两种内置数据源支持。

  • 基本的数据源：Streaming API中直接提供的数据源。例如文件系统和套接字连接。
  • 高级数据源：Kafka、Flume、Kinesis等数据源，这种高级数据源需要提供额外Maven依赖。
    注意：因为Receiver接收器是运行在Executor上的，并且每个Receiver运行需要一个Core，如果指定多个Receiver接收器同时接收数据，那么就要给Executor分配足够的Core，因为Streaming程序运行的时候，Excutor是长期运行的。并且在本地测试的时候，MasterURl不要指定local或者local[1]，因为Streaming应用程序最少需要两个线程，一个用于接受数据，一个用于处理接受到的数据，如果你指定一个线程，那么程序只会运行Receiver接收器，而不会有其他的线程去处理接收到的数据。

输入DStream的基本数据源

利用Streaming API来创建DSteam，它提供了通过文件（任何兼容HDFS文件系统的分布式文件系统）创建DStream的输入源。以下API方法的功能就是SparkStreaming将监控dataDirectory这个目录，并且处理该目录中任何创建的文件（不支持监控嵌套目录）。

package com.lyz.streaming.datasource.genericimport org.apache.hadoop.io.{LongWritable, Text}
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat
import org.apache.spark.{SparkConf, rdd}
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}object FileStreamTest {def main(args: Array[String]): Unit = {//初始化SparkConfval conf = new SparkConf().setMaster("local[2]").setAppName("Streamingtest")//设置序列化类型.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")val streaming = new StreamingContext(conf, Seconds(2))/** * LongWritable：读取HDFS文件返回的偏移量，也就是一行文本所在文件的位置* Text：读取HDFS文件的返回Text类型，也就是Java中的String类型* TextInputFormat：指定读取HDFS文件的类型*/val stream: InputDStream[(LongWritable, Text)] = streaming.fileStream[LongWritable, Text, TextInputFormat]("hdfs://xxxxx:8020/spark/data")stream.foreachRDD(rdd => {rdd.foreach(res => {println(res._2)})})streaming.start()streaming.awaitTermination()}
}

注意：

  • 监控的目录中的文件格式要一样
  • 目录中创建文件的方式只能是移动或者重命名
  • 移动后更改文件，SparkStreaming是不会读取新的数据的，因为监控文件以后，在SparkStreaming中的数据是不可变的了。
    对于简单的文件，可以利用streamingContext.textFileStream(dataDiretory)方法读取文件，文件流不需要Receiver接收器，不需要单独配备Core。

自定义Receiver接收器

SparkStreaming可以接收任何数据源发来的数据流，如果数据源超出了SparkStreaming内置的数据源限制，那么开发人员可以根据实际的业务自定义Receiver数据接收器。需要注意的是自定义接收器只能支持Java和Scala语言。


/**** 自定义一个类，继承抽象类Receiver，并且要根据业务指定Receiver的泛型，这个泛型就是接收到的数据后给* SparkStreaming的数据类型* Receiver的构造参数指定就是接受到数据后的本地化策略*/
class CustomReceiver(host: String, port: Int) extends Receiver[String](StorageLevel.MEMORY_AND_DISK_2) {def onStart() {/*** 因为onStart的方法和onStop方法不能无限制的阻塞，所以其中一方需要开启一个单独的线程。*/new Thread("Socket Receiver") {override def run() {receive()}}.start()}def onStop() {// There is nothing much to do as the thread calling receive()// is designed to stop by itself if isStopped() returns false}/*** 具体的数据输入源，需要对异常进行捕获，防止接收器故障*/private def receive() {var socket: Socket = nullvar userInput: String = ""try {// Connect to host:portsocket = new Socket(host, port)// Until stopped or connection broken continue readingval reader = new BufferedReader(new InputStreamReader(socket.getInputStream(), StandardCharsets.UTF_8))userInput = reader.readLine()while (!isStopped && userInput != null) {userInput = reader.readLine()this.store(userInput)}reader.close()socket.close()/*** restart方法的执行流程就是首先异步调用调用onStop方法之后，在延期调用onStart方法，** stop<exception>:将会调用onStop方法停止Receiver接收器。** reportError<exception>：就是停止也不重启Reveiver接收器，将错误信息报告给Driver程序*/restart("Trying to connect again")} catch {case e: java.net.ConnectException =>// restart if could not connect to serverrestart("Error connecting to " + host + ":" + port, e)case t: Throwable =>// restart if there is any other errorrestart("Error receiving data", t)}}
}

高级数据源之Kafka

Apache Kafka是基于生产者-消费者模型的一个分布式的消息服务。SparkStreaming和Kafka集成，提供了简单的并行性，Kafka分区和Spark分区是1.1对应关系。

Kafka与SparkStreaming对应关系

spark-streaming-kafka-0-8spark-streaming-kafka-0-10
Broker Version0.8.2.1 or higher0.10.0 or higher

添加Maven依赖

<dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming-kafka-0-10_2.11</artifactId><version>2.2.2</version>
</dependency>

通过Kafka直接创建DStream，需要注意的是，kafka导入的包是org.apache.spark.streaming.kafka010。

package com.lyz.streaming.datasource.advanceimport org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies._
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribeobject KafkaDataSourceTest {def main(args: Array[String]): Unit = {val kafkaParams = Map[String, Object]("bootstrap.servers" -> "localhost:9092", //指定Kafka的集群地址"key.deserializer" -> classOf[StringDeserializer], //指定key的反序列化器"value.deserializer" -> classOf[StringDeserializer], //指定值的反序列化器"group.id" -> "", //consumer的分组id"auto.offset.reset" -> "latest", //从新定义消费者以后，不从头消费分区里的数据，只消费定义消费者以后的数据"enable.auto.commit" -> (false: java.lang.Boolean) //是否自动提交offsets，也就是更新kafka里的offset，表示已经被消费过了)//定义消费主题topicval topics = Array("topic1", "topic2")val sparkConf: SparkConf = new SparkConf().setMaster("local[3]").setAppName("KafkaDatasourceTest")val streamingContext: StreamingContext = new StreamingContext(sparkConf, Seconds(2))/*** PreferConsistent：分区策略 ---->在可用的Executor之间均匀分配分区* PreferBrokers：分区策略 ---->只有当执行程序与Kafka代理程序位于相同的节点时，才可以使用。* PreferFixed：分区策略 ---->分区之间的负载偏差比较大，就该用这个分区策略* Subscribe：消费策略 ---->消费固定主题上的消息* SubscribePattern：消费策略 ---->消费正则匹配到的主题上的消息* Assign：消费策略 ---->消费固定分区集合上消息* ConsumerRecord：包含了主题名、分区号、分区记录的偏移量、具体的值*/val kafkaStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](streamingContext,PreferConsistent,Subscribe[String, String](topics, kafkaParams))kafkaStream.map(record => (record.key(), record.value()))}
}

注意：如果Spark批处理的时间间隔大于Kafka默认的心跳回话超时时间（30s），请适当增减Kafka里heartbeat.internal.ms和session.timeout.ms对于大于五分钟的批次,需要改变，group.max.session.timeout.ms。有关Kafka具体的配置参数请参考http://kafka.apache.org/documentation.html#newconsumerconfigs

LocationStrategies位置策略

新的Kafka消费者API能够预先将分区上消息缓存到缓冲区中。处于性能原因，Spark会将消息保留在Executor上，而不是每个批次都重新创建它们。并且能够在消费者的宿主机上适当的调度分区。如果需要禁用消费者缓存可以配置spark.streaming.kafka.consumer.cahce.enable为false。

object KafkaDataSourceTest {def main(args: Array[String]): Unit = {val kafkaParams = ...val topics = ...val sparkConf: SparkConf = ....val streamingContext: StreamingContext =....//禁用消费者预先缓存分区上的消息sparkConf.set("spark.streaming.kafka.consumer.cahce.enable", "false")val kafkaStream: InputDStream[ConsumerRecord[String, String]] = ...}
}

大多数情况下你可以使用LocationStrategies.PreferConsistent来指定Executor之间的均匀的分区策略。如果你的Executor与Kafka代理位于同一台宿主机上，那么你就可以使用LocationStrategies.PreferBrokers，这将优先为该分区上的Kafka Leader安排分区。如果分区分区的负载偏移差过大，那么就可以选择LocationStrategies.PreferFixed来手动安排分区。

消费者的默认缓存大小是64，如果你希望并行处理超过64*Executor大小的数据，那么你就需要修改spark.streaming.kafka.consumer.cache.maxCapacity参数。

object KafkaDataSourceTest {def main(args: Array[String]): Unit = {val kafkaParams = ...val topics = ...val sparkConf: SparkConf = ....val streamingContext: StreamingContext =....//配置SparkStreaming处理最大数据的大小sparkConf.set("spark.streaming.kafka.consumer.cache.maxCapacity", "80")val kafkaStream: InputDStream[ConsumerRecord[String, String]] = ...}
}

ConsumerStrategies消费策略

新的Kafka消费者API可以从不同的主题上消费消息。ConsumerStrategies提供了一个抽象，它可以指定具体消费哪个主题的策略。

ConsumerStrategies.Subscribe指定的是消费固定主题集合上的消息。Consumer.Strategies.SubscribePattern指定的是消费正则匹配到的主题上的消息。Assign指定了消费具体的分区集合上的消息。

基于Kafka创建RDD

如果你有一个更适合批处理的实例的话，你可以根据offset的偏移量范围来创建RDD

package com.lyz.streaming.datasource.advance
import java.util
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010.{KafkaUtils, LocationStrategies, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}object CreateRddByOffset {def main(args: Array[String]): Unit = {//定义一个Java Map 初始化连接kafka参数val kafkaParams = new util.HashMap[String, Object]()kafkaParams.put("bootstrap.servers", "localhost:9092")kafkaParams.put("key.deserializer", classOf[StringDeserializer])kafkaParams.put("value.deserializer", classOf[StringDeserializer])kafkaParams.put("group.id", "")kafkaParams.put("auto.offset.reset", "latest")kafkaParams.put("enable.auto.commit", false: java.lang.Boolean)val offsetRangs = Array(//参数解释依次是 主题名字、分区索引、offset开始、offset结束OffsetRange("test", 0, 0, 100),OffsetRange("test", 1, 0, 100),OffsetRange("test", 2, 0, 100))val sparkConf: SparkConf = new SparkConf().setMaster("local[3]").setAppName("KafkaDatasourceTest")val streamingContext: StreamingContext = new StreamingContext(sparkConf, Seconds(2))/** * 、根据分区的偏移量来创建RDD，* kafkaParams:这个参数是Java的HashMap*/val rdd = KafkaUtils.createRDD[String, String](streamingContext.sparkContext, kafkaParams, offsetRangs, LocationStrategies.PreferConsistent)}
}

保证数据完整性之获取偏移量并存储

当我们接受到Kafka上的数据以后，会将数据转换成DStream，而DStream底层其实就是时间片上的RDD。我们都知道RDD是以分区的形式散落到每个分区上的，由于RDD上的分区与Kafka上的分区是一一对应的。如果我们操作RDD的时候出现了Shuffle操作，那么RDD的分区与Kafka的分区之间的一一映射关系不会被保留，这样如果中间出现了操作失败的话，由于默认情况下Kafka的消费者会定期提交偏移量，那么消费者就不会再次消费到之前的数据，这样就影响数据的可靠性和准确性。我们可以获取RDD分区对应Kafka分区上的偏移量并保存下来，当处理Shuffle失败的时候，我们就可以按照偏移量重新去Kafka分区上重新加载数据，这样就保证了数据的可靠性和准确性。

注意：想要保证数据的可靠性和完整性，我们要关闭消费者自动定期的提交偏移量。 设置Kafka的enable.auto.commit参数 为 (false: java.lang.Boolean)

package com.lyz.streaming.datasource.advanceobject KafkaDataSourceTest {def main(args: Array[String]): Unit = {val kafkaParams = ...//定义消费主题topicval topics = Array("topic1", "topic2")val sparkConf: SparkConf =...val streamingContext: StreamingContext = ...val kafkaStream: InputDStream[ConsumerRecord[String, String]] = ...//获取RDD分区对应Kafka上分区的偏移量kafkaStream.foreachRDD(rdd => {val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRangesrdd.foreachPartition(partion => {val range = offsetRanges(TaskContext.get().partitionId())val topic: String = range.topic //该RDD对应分区上的Kafka主题val fromOffset: Long = range.fromOffset //该RDD对应分区上的数据在Kafka上偏移量开始的位置val untilOffset: Long = range.untilOffset //该RDD对应分区上的数据在Kafka上偏移量结束的位置val partition: Int = range.partition //该RDD对应分区上的数据对应kafka分区的位置})})}
}

保证数据完整性之Kafka本身存储偏移量

Kafka有一个新的偏移量API，用于在特殊的主题中存储偏移量。默认情况下消费者会定期自动提交偏移量，这样消费者消费了Kakfa上的数据以后可能没有结果输出或者出现处理异常，影响了数据的完整性，这就是为什么我们要设置enable.auto.commit为false的原因。但是使用commitAsync API后就可以解决这个问题

package com.lyz.streaming.datasource.advanceobject CommitAsyncTest {def main(args: Array[String]): Unit = {val kafkaParams = ...//定义消费主题topicval topics = Array("topic1", "topic2")val sparkConf: SparkConf = ...val streamingContext: StreamingContext = ...val kafkaStream: InputDStream[ConsumerRecord[String, String]] = ...kafkaStream.foreachRDD(rdd => {//这个必须处理RDD上的数据之前调用val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRangesrdd.foreachPartition(partion => {//处理你的业务逻辑})/*** 这个API必须在处理RDD数据得到结果之后调用* commitAsync这个方法是线程安全的*/kafkaStream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)})}
}

保证数据完整性之自定义存储偏移量

为了SparkStreaming业务的事务性，我们最好把业务计算结果和offset同时进行存储，这样可以保证要么都成功，要么都失败，这样就形成了一个原子操作。由于一个业务有可能是不同分区上、不同主题上、不同的offset上的数据，这样我们应该利用Assign来指定消费具体的分区集合上的消息。一下代码我是将offset信息保存在Hbase中，每次处理数据首先从Hbase中查询出上一次offset结束的位置开始消费。

package com.lyz.streaming.datasource.advanceimport java.utilimport org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.client._
import org.apache.hadoop.hbase.mapreduce.TableInputFormat
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.hbase.{HBaseConfiguration, HColumnDescriptor, HTableDescriptor, TableName}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferFixed
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.{Seconds, StreamingContext}/*** 自定义消费的主题和分区测试类*/
object AysncTest {def main(args: Array[String]): Unit = {//Hbase表名val topic_partition_offset = "topic_partition_offset"//Hbase表的列簇名val hbase_comsumer_offset = "hbase_comsumer_offset"//Kafka出书画参数val kafkaParams = Map[String, Object]("bootstrap.servers" -> "192.168.101.215:9092", //指定Kafka的集群地址"key.deserializer" -> classOf[StringDeserializer], //指定key的反序列化器"value.deserializer" -> classOf[StringDeserializer], //指定值的反序列化器"group.id" -> "0", //consumer的分组id，相同组的消费者共同消费一个topic上的数据，每一个消费者消费单独partition数据"auto.offset.reset" -> "latest", //从新定义消费者以后，不从头消费分区里的数据，只消费定义消费者以后的数据"enable.auto.commit" -> (false: java.lang.Boolean) //是否自动提交offsets，也就是更新kafka里的offset，表示已经被消费过了)//定义消费主题topicval topics = Array("topicname", "topicname1")//初始化SparkConfval sparkConf: SparkConf = new SparkConf().setMaster("local[3]").setAppName("KafkaDatasourceTest")//这个配置是解决Spark读取Hbase抛java.io.NotSerializableException异常的问题.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")//初始化SparkStreamingval streamingContext: StreamingContext = new StreamingContext(sparkConf, Seconds(5))//获取Hbase配置val hbaseConf: Configuration = HBaseConfiguration.create()val hbaseConn: Connection = ConnectionFactory.createConnection(hbaseConf) //根据Hbase配置初始化Hbase连接val admin: Admin = hbaseConn.getAdminval tableName: TableName = TableName.valueOf(hbase_comsumer_offset)val isExsist: Boolean = hbaseConn.getAdmin.tableExists(tableName)var stream: InputDStream[ConsumerRecord[String, String]] = null//判断Hbase是否存在offset表，没有就从topic头开始消费if (!isExsist) {//定义一个Hbase表val hTable = new HTableDescriptor(TableName.valueOf(hbase_comsumer_offset))//定义Hbase表的列簇hTable.addFamily(new HColumnDescriptor(Bytes.toBytes(topic_partition_offset)))//创建Hbase表admin.createTable(hTable)stream = KafkaUtils.createDirectStream[String, String](streamingContext,PreferFixed(Map[TopicPartition, String]()),Subscribe[String, String](topics, kafkaParams))} else { //如果有就从上一次消费到的offset开始消费//配置spark读取Hbase的哪个表hbaseConf.set(TableInputFormat.INPUT_TABLE, tableName.toString)//配置spark读取Hbase表中的那个列簇hbaseConf.set(TableInputFormat.SCAN_COLUMN_FAMILY, topic_partition_offset)//Spark读取Hbase上的数据val hbaseRdd = streamingContext.sparkContext.newAPIHadoopRDD(hbaseConf, classOf[TableInputFormat], classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable], classOf[org.apache.hadoop.hbase.client.Result])//获取Hbase中offset的信息val fromOffset: Map[TopicPartition, Long] = hbaseRdd.map(rdd => {val res: Result = rdd._2val topic: String = Bytes.toString(res.getValue(Bytes.toBytes(topic_partition_offset), Bytes.toBytes("topic")))val partition: Int = Bytes.toInt(res.getValue(Bytes.toBytes(topic_partition_offset), Bytes.toBytes("partition")))val untilOffset: Long = Bytes.toLong(res.getValue(Bytes.toBytes(topic_partition_offset), Bytes.toBytes("untilOffset")))new TopicPartition(topic, partition) -> untilOffset}).collect().toMap//接着从上一次读取的到offset开始读取Kafka上的数据stream = KafkaUtils.createDirectStream[String, String](streamingContext,LocationStrategies.PreferConsistent,ConsumerStrategies.Assign[String, String](fromOffset.keys.toList, kafkaParams, fromOffset))}stream.foreachRDD(rdd => {//这个必须处理RDD上的数据之前调用val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges//将offset存储到Hbase上val t: Table = hbaseConn.getTable(tableName)//定义Put容器val putList = new util.ArrayList[Put]//更新Hbase上的Offerset信息for (offset <- offsetRanges) {val put = new Put(Bytes.toBytes(offset.topic + "_" + kafkaParams.get("group.id")))put.addColumn(Bytes.toBytes(topic_partition_offset), Bytes.toBytes("topic"), Bytes.toBytes(offset.topic))put.addColumn(Bytes.toBytes(topic_partition_offset), Bytes.toBytes("partition"), Bytes.toBytes(offset.partition))put.addColumn(Bytes.toBytes(topic_partition_offset), Bytes.toBytes("untilOffset"), Bytes.toBytes(offset.untilOffset))putList.add(put)}t.put(putList)rdd.foreachPartition(partion => {//你自己的处理业务逻辑})})admin.flush(tableName)streamingContext.start()streamingContext.awaitTermination()}
}

在Spark读取Hbase上的数据的时候，想要精确点，你可以利用Hbase的Configuration配置具体的参数，如下所示

public static final String INPUT_TABLE = "hbase.mapreduce.inputtable";    
private static final String SPLIT_TABLE = "hbase.mapreduce.splittable";    
public static final String SCAN = "hbase.mapreduce.scan";  
public static final String SCAN_ROW_START = "hbase.mapreduce.scan.row.start";   
public static final String SCAN_ROW_STOP = "hbase.mapreduce.scan.row.stop";    
public static final String SCAN_COLUMN_FAMILY = "hbase.mapreduce.scan.column.family";    
public static final String SCAN_COLUMNS = "hbase.mapreduce.scan.columns";   
public static final String SCAN_TIMESTAMP = "hbase.mapreduce.scan.timestamp";    
public static final String SCAN_TIMERANGE_START = "hbase.mapreduce.scan.timerange.start";    
public static final String SCAN_TIMERANGE_END = "hbase.mapreduce.scan.timerange.end";    
public static final String SCAN_MAXVERSIONS = "hbase.mapreduce.scan.maxversions";    
public static final String SCAN_CACHEBLOCKS = "hbase.mapreduce.scan.cacheblocks";    
public static final String SCAN_CACHEDROWS = "hbase.mapreduce.scan.cachedrows";    
public static final String SCAN_BATCHSIZE = "hbase.mapreduce.scan.batchsize";    
public static final String SHUFFLE_MAPS = "hbase.mapreduce.inputtable.shufflemaps";

高级数据源之Flume

Apache Flume是一个分布式、可靠的服务，它能够高效收集、聚合和移动大量日志。这里我们将介绍怎样配置Flume和SparkStreaming从Flume上接受数据。SparkStreaming接受Flume上的数据有两种方法。

Flume推送数据给Streaming方式

Flume被设计用来推送数据的，这种方式，SparkStreaming作为一个Avro数据池接受器来接受Flume推送过来的数据。这种方式设置器来比较容易，但是它的缺点就是不使用事物接收数据，并且可能数据量一大，Streaming消费不过来而造成节点的高负载，如果消费节点出现故障，那么数据就会丢失，造成数据的不完整性。

Flume的配置，Flume的具体配置请参考Flume学习总结

#定义sources、sinks、channels
a1.sources = r1
a1.sinks = k1
a1.channels = c1#定义sources的属性配置
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444#定义sinks的属性配置
a1.sinks.avroSink.type = avro
a1.sinks.avroSink.channel = memory
a1.sinks.avroSink.hostname = xxxxx
a1.sinks.avroSink.port = xxxx#定义channels属性配置
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100#利用channel连接source和sink
a1.sources.r1.channels=c1
a1.sinks.avroSink.channels=c1

SparkStreaming代码如下

package com.lyz.streaming.datasource.advanceimport org.apache.spark.{SparkConf, rdd}
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.flume.{FlumeUtils, SparkFlumeEvent}
import org.apache.spark.streaming.{Seconds, StreamingContext}object FlumeDataSourceTest {def main(args: Array[String]): Unit = {val sparkConf: SparkConf = new SparkConf().setMaster("local[3]").setAppName("KafkaDatasourceTest")val streamingContext: StreamingContext = new StreamingContext(sparkConf, Seconds(2))//作为avro数据池，来接受Flume推送过来的数据val stream: ReceiverInputDStream[SparkFlumeEvent] = FlumeUtils.createStream(streamingContext, "localhost", 44444)stream.foreachRDD(rdd => {rdd.collect().foreach(println(_))})streamingContext.start()streamingContext.awaitTermination()}
}

Streaming从Flume上拉取数据方式

这种方式Streaming自动从Flume的缓存池中拉取数据，Streaming通过事物从数据池中读取并复制数据，在收到事物完成的通知前，这些数据还会保留在数据池中。这样保证了数据的完整性。

由于这种方式使用的是第三方插件的Sink，它是Scala写的，所以我们要把这插件以及Scala库添加到Flume的类路径中。

插件以及Scala库的Maven

flume-sink -->
<dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming-flume-sink_2.11</artifactId><version>2.2.2</version>
</dependency><dependency><groupId>org.scala-lang</groupId><artifactId>scala-library</artifactId><version>2.11.8</version></dependency><dependency><groupId>org.apache.commons</groupId><artifactId>commons-lang3</artifactId><version>3.5</version>
</dependency>

Flume配置文件

#定义sources、sinks、channels
a1.sources = r1
a1.sinks = k1
a1.channels = c1#定义sources的属性配置
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444#定义sinks的属性配置
agent.sinks = spark
agent.sinks.spark.type = org.apache.spark.streaming.flume.sink.SparkSink
agent.sinks.spark.hostname = <hostname of the local machine>
agent.sinks.spark.port = <port to listen on for connection from Spark>
agent.sinks.spark.channel = memoryChannel#定义channels属性配置
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100#利用channel连接source和sink
a1.sources.r1.channels=c1
a1.sinks.avroSink.channels=c1

Streaming代码

package com.lyz.streaming.datasource.advanceimport org.apache.spark.{SparkConf, rdd}
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.flume.{FlumeUtils, SparkFlumeEvent}
import org.apache.spark.streaming.{Seconds, StreamingContext}object FlumeDataSourceTest {def main(args: Array[String]): Unit = {val sparkConf: SparkConf = new SparkConf().setMaster("local[3]").setAppName("KafkaDatasourceTest")val streamingContext: StreamingContext = new StreamingContext(sparkConf, Seconds(2))//利用Streaming拉取Flume数据池的数据方式val stream: ReceiverInputDStream[SparkFlumeEvent] = FlumeUtils.createPollingStream(streamingContext, "", 44444)stream.foreachRDD(rdd => {rdd.collect().foreach(println(_))})streamingContext.start()streamingContext.awaitTermination()}
}

这篇关于Spark Streaming（二十六）DStream基本数据源、高级数据源的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/753070。

相关文章

mysql中insert into的基本用法和一些示例

mysql中insert into的基本用法和一些示例

《mysql中insertinto的基本用法和一些示例》INSERTINTO用于向MySQL表插入新行,支持单行/多行及部分列插入,下面给大家介绍mysql中insertinto的基本用法和一些示例... 目录基本语法插入单行数据插入多行数据插入部分列的数据插入默认值注意事项在mysql中，INSERT I
阅读更多...
mapstruct中的@Mapper注解的基本用法

mapstruct中的@Mapper注解的基本用法

《mapstruct中的@Mapper注解的基本用法》在MapStruct中,@Mapper注解是核心注解之一,用于标记一个接口或抽象类为MapStruct的映射器（Mapper）,本文给大家介绍ma... 目录1. 基本用法2. 常用属性3. 高级用法4. 注意事项5. 总结6. 编译异常处理在MapSt
阅读更多...
MyBatis ResultMap 的基本用法示例详解

MyBatis ResultMap 的基本用法示例详解

《MyBatisResultMap的基本用法示例详解》在MyBatis中,resultMap用于定义数据库查询结果到Java对象属性的映射关系,本文给大家介绍MyBatisResultMap的基本... 目录MyBATis 中的 resultMap1. resultMap 的基本语法2. 简单的 resul
阅读更多...
Java 枚举的基本使用方法及实际使用场景

Java 枚举的基本使用方法及实际使用场景

《Java枚举的基本使用方法及实际使用场景》枚举是Java中一种特殊的类,用于定义一组固定的常量,枚举类型提供了更好的类型安全性和可读性,适用于需要定义一组有限且固定的值的场景,本文给大家介绍Jav... 目录一、什么是枚举？二、枚举的基本使用方法定义枚举三、实际使用场景代替常量状态机四、更多用法1.实现接
阅读更多...
git stash命令基本用法详解

git stash命令基本用法详解

《gitstash命令基本用法详解》gitstash是Git中一个非常有用的命令,它可以临时保存当前工作区的修改,让你可以切换到其他分支或者处理其他任务,而不需要提交这些还未完成的修改,这篇文章主要... 目录一、基本用法1. 保存当前修改（包括暂存区和工作区的内容）2. 查看保存了哪些 stash3. 恢
阅读更多...
Apache 高级配置实战之从连接保持到日志分析的完整指南

Apache 高级配置实战之从连接保持到日志分析的完整指南

《Apache高级配置实战之从连接保持到日志分析的完整指南》本文带你从连接保持优化开始,一路走到访问控制和日志管理,最后用AWStats来分析网站数据,对Apache配置日志分析相关知识感兴趣的朋友... 目录Apache 高级配置实战：从连接保持到日志分析的完整指南前言 一、Apache 连接保持 - 性
阅读更多...
MySQL基本查询示例总结

MySQL基本查询示例总结

《MySQL基本查询示例总结》：本文主要介绍MySQL基本查询示例总结,本文通过实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友参考下吧... 目录Create插入替换Retrieve(读取)select（确定列）where条件（确定行）null查询order by语句li
阅读更多...
mysql中的group by高级用法详解

mysql中的group by高级用法详解

《mysql中的groupby高级用法详解》MySQL中的GROUPBY是数据聚合分析的核心功能,主要用于将结果集按指定列分组,并结合聚合函数进行统计计算,本文给大家介绍mysql中的groupby... 目录一、基本语法与核心功能二、基础用法示例1. 单列分组统计2. 多列组合分组3. 与WHERE结合使
阅读更多...
PyTorch高级特性与性能优化方式

PyTorch高级特性与性能优化方式

《PyTorch高级特性与性能优化方式》：本文主要介绍PyTorch高级特性与性能优化方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一、自动化机制1.自动微分机制2.动态计算图二、性能优化1.内存管理2.GPU加速3.多GPU训练三、分布式训练1.分布式数据
阅读更多...
Spring Boot集成SLF4j从基础到高级实践(最新推荐)

Spring Boot集成SLF4j从基础到高级实践(最新推荐)

《SpringBoot集成SLF4j从基础到高级实践(最新推荐)》SLF4j(SimpleLoggingFacadeforJava)是一个日志门面（Facade）,不是具体的日志实现,这篇文章主要介... 目录一、日志框架概述与SLF4j简介1.1 为什么需要日志框架1.2 主流日志框架对比1.3 SLF4
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2