设为首页 加入收藏

TOP

streaming流式读取hdfs采坑记
2019-04-14 12:13:50 】 浏览:10
Tags:streaming 流式 读取 hdfs
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了


编程开发网
】【打印繁体】【投稿】【收藏】 【推荐】【举报】【评论】 【关闭】 【返回顶部
上一篇配置本地eclipse,在eclipse中操.. 下一篇本地文件上传至hdfs文件系统

评论

帐  号: 密码: (新用户注册)
验 证 码:
表  情:
内  容:

array(4) { ["type"]=> int(8) ["message"]=> string(24) "Undefined variable: jobs" ["file"]=> string(32) "/mnt/wp/cppentry/do/bencandy.php" ["line"]=> int(214) }