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。

相关文章

JavaScript中的高级调试方法全攻略指南

JavaScript中的高级调试方法全攻略指南

《JavaScript中的高级调试方法全攻略指南》什么是高级JavaScript调试技巧,它比console.log有何优势,如何使用断点调试定位问题,通过本文,我们将深入解答这些问题,带您从理论到实... 目录观点与案例结合观点1观点2观点3观点4观点5高级调试技巧详解实战案例断点调试：定位变量错误性能分
阅读更多...
从基础到高级详解Python数值格式化输出的完全指南

从基础到高级详解Python数值格式化输出的完全指南

《从基础到高级详解Python数值格式化输出的完全指南》在数据分析、金融计算和科学报告领域,数值格式化是提升可读性和专业性的关键技术,本文将深入解析Python中数值格式化输出的相关方法,感兴趣的小伙... 目录引言：数值格式化的核心价值一、基础格式化方法1.1 三种核心格式化方式对比1.2 基础格式化示例
阅读更多...
Python ORM神器之SQLAlchemy基本使用完全指南

Python ORM神器之SQLAlchemy基本使用完全指南

《PythonORM神器之SQLAlchemy基本使用完全指南》SQLAlchemy是Python主流ORM框架,通过对象化方式简化数据库操作,支持多数据库,提供引擎、会话、模型等核心组件,实现事务... 目录一、什么是SQLAlchemy？二、安装SQLAlchemy三、核心概念1. Engine（引擎）
阅读更多...
Android协程高级用法大全

Android协程高级用法大全

《Android协程高级用法大全》这篇文章给大家介绍Android协程高级用法大全,本文结合实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友跟随小编一起学习吧... 目录1️⃣ 协程作用域（CoroutineScope）与生命周期绑定Activity/Fragment 中手
阅读更多...
Python异步编程之await与asyncio基本用法详解

Python异步编程之await与asyncio基本用法详解

《Python异步编程之await与asyncio基本用法详解》在Python中,await和asyncio是异步编程的核心工具,用于高效处理I/O密集型任务（如网络请求、文件读写、数据库操作等）,接... 目录一、核心概念二、使用场景三、基本用法1. 定义协程2. 运行协程3. 并发执行多个任务四、关键
阅读更多...
Go语言连接MySQL数据库执行基本的增删改查

Go语言连接MySQL数据库执行基本的增删改查

《Go语言连接MySQL数据库执行基本的增删改查》在后端开发中,MySQL是最常用的关系型数据库之一,本文主要为大家详细介绍了如何使用Go连接MySQL数据库并执行基本的增删改查吧... 目录Go语言连接mysql数据库准备工作安装 MySQL 驱动代码实现运行结果注意事项Go语言执行基本的增删改查准备工作
阅读更多...
深度解析Python yfinance的核心功能和高级用法

深度解析Python yfinance的核心功能和高级用法

《深度解析Pythonyfinance的核心功能和高级用法》yfinance是一个功能强大且易于使用的Python库,用于从YahooFinance获取金融数据,本教程将深入探讨yfinance的核... 目录yfinance 深度解析教程 (python)1. 简介与安装1.1 什么是 yfinance?
阅读更多...
MySQL数据类型与表操作全指南( 从基础到高级实践)

MySQL数据类型与表操作全指南( 从基础到高级实践)

《MySQL数据类型与表操作全指南(从基础到高级实践)》本文详解MySQL数据类型分类（数值、日期/时间、字符串）及表操作（创建、修改、维护）,涵盖优化技巧如数据类型选择、备份、分区,强调规范设计与... 目录mysql数据类型详解数值类型日期时间类型字符串类型表操作全解析创建表修改表结构添加列修改列删除列
阅读更多...
java实现多数据源切换方式

java实现多数据源切换方式

《java实现多数据源切换方式》本文介绍实现多数据源切换的四步方法：导入依赖、配置文件、启动类注解、使用@DS标记mapper和服务层,通过注解实现数据源动态切换,适用于实际开发中的多数据源场景... 目录一、导入依赖二、配置文件三、在启动类上配置四、在需要切换数据源的类上、方法上使用@DS注解结论一、导入
阅读更多...
DNS查询的利器! linux的dig命令基本用法详解

DNS查询的利器! linux的dig命令基本用法详解

《DNS查询的利器!linux的dig命令基本用法详解》dig命令可以查询各种类型DNS记录信息，下面我们将通过实际示例和dig命令常用参数来详细说明如何使用dig实用程序... dig（Domain Information Groper）是一款功能强大的 linux 命令行实用程序，通过查询名称服务器并输
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2