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。

相关文章

C++中使用vector存储并遍历数据的基本步骤

C++中使用vector存储并遍历数据的基本步骤

《C++中使用vector存储并遍历数据的基本步骤》C++标准模板库（STL）提供了多种容器类型,包括顺序容器、关联容器、无序关联容器和容器适配器,每种容器都有其特定的用途和特性,：本文主要介绍C... 目录（1）容器及简要描述‌php顺序容器‌‌关联容器‌‌无序关联容器‌（基于哈希表）：‌容器适配器‌：（
阅读更多...
使用Python进行文件读写操作的基本方法

使用Python进行文件读写操作的基本方法

《使用Python进行文件读写操作的基本方法》今天的内容来介绍Python中进行文件读写操作的方法,这在学习Python时是必不可少的技术点,希望可以帮助到正在学习python的小伙伴,以下是Pyth... 目录一、文件读取：二、文件写入：三、文件追加：四、文件读写的二进制模式：五、使用 json 模块读写
阅读更多...
SpringBoot操作spark处理hdfs文件的操作方法

SpringBoot操作spark处理hdfs文件的操作方法

《SpringBoot操作spark处理hdfs文件的操作方法》本文介绍了如何使用SpringBoot操作Spark处理HDFS文件,包括导入依赖、配置Spark信息、编写Controller和Ser... 目录SpringBoot操作spark处理hdfs文件1、导入依赖2、配置spark信息3、cont
阅读更多...
Python中列表的高级索引技巧分享

Python中列表的高级索引技巧分享

《Python中列表的高级索引技巧分享》列表是Python中最常用的数据结构之一,它允许你存储多个元素,并且可以通过索引来访问这些元素,本文将带你深入了解Python列表的高级索引技巧,希望对... 目录1.基本索引2.切片3.负数索引切片4.步长5.多维列表6.列表解析7.切片赋值8.删除元素9.反转列表
阅读更多...
Spring Boot实现多数据源连接和切换的解决方案

Spring Boot实现多数据源连接和切换的解决方案

《SpringBoot实现多数据源连接和切换的解决方案》文章介绍了在SpringBoot中实现多数据源连接和切换的几种方案,并详细描述了一个使用AbstractRoutingDataSource的实... 目录前言一、多数据源配置与切换方案二、实现步骤总结前言在 Spring Boot 中实现多数据源连接
阅读更多...
正则表达式高级应用与性能优化记录

正则表达式高级应用与性能优化记录

《正则表达式高级应用与性能优化记录》本文介绍了正则表达式的高级应用和性能优化技巧,包括文本拆分、合并、XML/HTML解析、数据分析、以及性能优化方法,通过这些技巧,可以更高效地利用正则表达式进行复杂... 目录第6章：正则表达式的高级应用6.1 模式匹配与文本处理6.1.1 文本拆分6.1.2 文本合并6
阅读更多...
基本知识点

基本知识点

1、c++的输入加上ios::sync_with_stdio(false);  等价于 c的输入，读取速度会加快（但是在字符串的题里面和容易出现问题） 2、lower_bound()和upper_bound() iterator lower_bound( const key_type &key ): 返回一个迭代器，指向键值>= key的第一个元素。 iterator upper_bou
阅读更多...
【IPV6从入门到起飞】5-1 IPV6+Home Assistant(搭建基本环境)

【IPV6从入门到起飞】5-1 IPV6+Home Assistant(搭建基本环境)

【IPV6从入门到起飞】5-1 IPV6+Home Assistant #搭建基本环境 1 背景2 docker下载 hass3 创建容器4 浏览器访问 hass5 手机APP远程访问hass6 更多玩法 1 背景 既然电脑可以IPV6入站，手机流量可以访问IPV6网络的服务，为什么不在电脑搭建Home Assistant（hass），来控制你的设备呢？@智能家居 @万物互联
阅读更多...
系统架构师考试学习笔记第三篇——架构设计高级知识（20）通信系统架构设计理论与实践

系统架构师考试学习笔记第三篇——架构设计高级知识（20）通信系统架构设计理论与实践

本章知识考点：         第20课时主要学习通信系统架构设计的理论和工作中的实践。根据新版考试大纲,本课时知识点会涉及案例分析题(25分),而在历年考试中,案例题对该部分内容的考查并不多,虽在综合知识选择题目中经常考查,但分值也不高。本课时内容侧重于对知识点的记忆和理解,按照以往的出题规律,通信系统架构设计基础知识点多来源于教材内的基础网络设备、网络架构和教材外最新时事热点技术。本课时知识
阅读更多...
C 语言的基本数据类型

C 语言的基本数据类型

C 语言的基本数据类型 注：本文面向 C 语言初学者，如果你是熟手，那就不用看了。 有人问我，char、short、int、long、float、double 等这些关键字到底是什么意思，如果说他们是数据类型的话，那么为啥有这么多数据类型呢？ 如果写了一句： int a; 那么执行的时候在内存中会有什么变化呢？ 橡皮泥大家都玩过吧，一般你买橡皮泥的时候，店家会赠送一些模板。 上
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2