package rockerMQ


import org.apache.spark.sql.SparkSession
import org.apache.spark.{SparkConf, SparkContext, sql}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * @Auther: sss
  * @Date: 2018/11/26 10:05
  * @Description:Streaming接收hdfsDemo
  */
object Demo02 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName(this.getClass.getSimpleName).setMaster("local[*]")
    val ssc = new StreamingContext(conf, Seconds(5))
    val session = SparkSession.builder().config(conf).getOrCreate()
    import session.implicits._
    val value = ssc.textFileStream("hdfs://192.168.xx.xx:9000/tmp/cxbtest")

    value.foreachRDD(rdd => {
     val df= rdd.map(row => {
        val name = row.split(" ", row.length)(0)
        val name1 = row.split(" ", row.length)(1)
        val name2 = row.split(" ", row.length)(2)
        Test(name, name1, name2)
      }).toDF()
      df.createOrReplaceTempView("tmp")
      session.sql("select name,name1,name2 from tmp where name='a' ").show()
    })

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

case class Test(name: String, name1: String, name2: String)

代码如上，自己测试写的

发现流式读取hdfs有个大坑，公司是用rocketMQ作为消息中间件，实时将数据接到hdfs上，因为不会

用sparkStreaming整合mq，就用这种方式来接，发现死活接收不到hdfs的流数据，自己在hdfs上创建目录然后再往里面上传文本发现代码是好使的，最后猜想原因是

自己往hdfs上传的文件或是文本都是几KB，而hdfs接收mq的数据都是几GB为一个文本，感觉流式读取hdfs是以

一个个文本为读取的批次？？ 不是没有读到文本，而是一个文本形成的时间太久了，hdfs还没形成一个文本，我这边就给kill了