分享一下spark streaming与flume集成的scala代码。

object LogicHandle {
  def main(args: Array[String]) {
    //添加这个不会报执行错误
    val path = new File(".").getCanonicalPath()
    System.getProperties().put("hadoop.home.dir", path);
    new File("./bin").mkdirs();
    new File("./bin/winutils.exe").createNewFile();

    //val sparkConf = new SparkConf().setAppName("SensorRealTime").setMaster("local[2]")
    val sparkConf = new SparkConf().setAppName("SensorRealTime")

    val ssc = new StreamingContext(sparkConf, Seconds(20))

    val hostname = "localhost"
    val port = 2345
    val storageLevel = StorageLevel.MEMORY_ONLY
    val flumeStream = FlumeUtils.createStream(ssc, hostname, port, storageLevel)

    val lhc = new LogicHandleClass();

    //日志格式化模板
    val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    val sdfHour = new SimpleDateFormat("HH");
    val sdfMinute = new SimpleDateFormat("mm")

    //存储数据的hash对象  key/value存储  根据文档规则，使用各统计指标的key/value
    var redisMap = new HashMap[String, String]

　　
　　　　
    flumeStream.foreachRDD(rdd => {
      val events = rdd.collect()
      //println("event count:" + events.length)
      var i = 1
      for (event <- events) {
        val sensorInfo = new String(event.event.getBody.array()) //单行记录
        //单行记录格式化
        val arrayFileds = sensorInfo.split(",")
        if (arrayFileds.length == 6) {
          val shopId = arrayFileds(0) //店内编号

          val floorId = shopId.substring(0, 5) //楼层编号
          val mac = arrayFileds(1)
          val ts = arrayFileds(2).toLong //时间戳
          val time = sdf.format(ts * 1000)
          var hour = sdfHour.format(ts * 1000)
          var minute = sdfMinute.format(ts * 1000)
          var allMinute = hour.toInt * 60 + minute.toInt

          val x = arrayFileds(3)
          val y = arrayFileds(4)
          val level = arrayFileds(5)

          //后边就是我的业务代码了，省略了
        }
      }

      //存储至redis中
      lhc.SetAll(redisMap)
    })

    ssc.start()
    ssc.awaitTermination()
  }
}