streaming流式读取hdfs采坑记

我会带着你远行 2022-04-13 04:22 244阅读 0赞

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了