输入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.generic

import 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 = {
    //初始化SparkConf
    val 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 = null
    var userInput: String = ""
    try {
      // Connect to host:port
      socket = new Socket(host, port)

      // Until stopped or connection broken continue reading
      val 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 server
        restart("Error connecting to " + host + ":" + port, e)
      case t: Throwable =>
        // restart if there is any other error
        restart("Error receiving data", t)
    }
  }
}

高级数据源之Kafka

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

Kafka与SparkStreaming对应关系

	spark-streaming-kafka-0-8	spark-streaming-kafka-0-10
Broker Version	0.8.2.1 or higher	0.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.advance

import 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.Subscribe

object 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，表示已经被消费过了
    )

    //定义消费主题topic
    val 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.advance

object KafkaDataSourceTest {
  def main(args: Array[String]): Unit = {
    val kafkaParams = ...
    //定义消费主题topic
    val 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].offsetRanges
	      rdd.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.advance

object CommitAsyncTest {
  def main(args: Array[String]): Unit = {

    val kafkaParams = ...

    //定义消费主题topic
    val 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].offsetRanges

      rdd.foreachPartition(partion => {
        //处理你的业务逻辑
      })
      
      /**
        * 这个API必须在处理RDD数据得到结果之后调用
        * commitAsync这个方法是线程安全的
        */
     kafkaStream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
    })
  }
}

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

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

package com.lyz.streaming.datasource.advance

import java.util

import 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，表示已经被消费过了
    )

    //定义消费主题topic
    val topics = Array("topicname", "topicname1")

    //初始化SparkConf
    val sparkConf: SparkConf = new SparkConf().setMaster("local[3]").setAppName("KafkaDatasourceTest")
      //这个配置是解决Spark读取Hbase抛java.io.NotSerializableException异常的问题
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

    //初始化SparkStreaming
    val 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.getAdmin
    val 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._2
        val 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.advance

import 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.advance

import 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()
  }
}