`
qindongliang1922
  • 浏览: 2146945 次
  • 性别: Icon_minigender_1
  • 来自: 北京
博客专栏
7265517b-f87e-3137-b62c-5c6e30e26109
证道Lucene4
浏览量:116311
097be4a0-491e-39c0-89ff-3456fadf8262
证道Hadoop
浏览量:124580
41c37529-f6d8-32e4-8563-3b42b2712a50
证道shell编程
浏览量:58448
43832365-bc15-3f5d-b3cd-c9161722a70c
ELK修真
浏览量:70345
社区版块
存档分类
最新评论

如何在Scala中读取Hadoop集群上的gz压缩文件

阅读更多


存在Hadoop集群上的文件,大部分都会经过压缩,如果是压缩后的文件,我们直接在应用程序中如何读取里面的数据?答案是肯定的,但是比普通的文本读取要稍微复杂一点,需要使用到Hadoop的压缩工具类支持,比如处理gz,snappy,lzo,bz压缩的,前提是首先我们的Hadoop集群得支持上面提到的各种压缩文件。

本次就给出一个读取gz压缩文件的例子核心代码:



  def readHdfsWriteKafkaByDate(fs:FileSystem,date:String,conf:Configuration,topic:String,finishTimeStamp:Long):Unit={
  
  //访问hdfs文件,只读取gz结尾的压缩文件,如果是.tmp结尾的不会读取
    val path=new Path("/collect_data/userlog/"+date+"/log*.gz")
    //实例化压缩工厂编码类
    val factory = new CompressionCodecFactory(conf)
    //读取通配路径
    val items=fs.globStatus(path)
    var count=0
    //遍历每一个路径文件
    items.foreach(f=>{
    //打印全路径
      println(f.getPath)
      //通过全路径获取其编码
      val codec = factory.getCodec(f.getPath())//获取编码
      //读取成数据流
      var  stream:InputStream = null;
      if(codec!=null){
      //如果编码识别直接从编码创建输入流
        stream = codec.createInputStream(fs.open(f.getPath()))
      }else{
      //如果不识别则直接打开
        stream = fs.open(f.getPath())
      }
      val writer=new StringWriter()
      //将字节流转成字符串流
      IOUtils.copy(stream,writer,"UTF-8")
      //得到字符串内容
      val raw=writer.toString
      //根据字符串内容split出所有的行数据,至此解压数据完毕
      val raw_array=raw.split("\n")
      //遍历数据      
      raw_array.foreach(line=>{

        val array = line.split("--",2) //拆分数组
        val map = JSON.parseObject(array(1)).asScala
        val userId = map.get("userId").getOrElse("").asInstanceOf[String] //为空为非法数据
        val time = map.get("time").getOrElse("") //为空为非法数据
        if(StringUtils.isNotEmpty(userId)&&(time+"").toLong<=finishTimeStamp){//只有数据
          pushToKafka(topic,userId,line)
          count=count+1
        }

      })

    })

  }



压缩和解压模块用的工具包是apache-commons下面的类:

import org.apache.commons.io.IOUtils
import org.apache.commons.lang.StringUtils


如果想在Windows上调试,可以直接设置HDFS的地址即可
-     val conf = new Configuration()//获取hadoop的conf
//    conf.set("fs.defaultFS","hdfs://192.168.10.14:8020/")//windows上调试用




至此数据已经解压并读取完毕,其实并不是很复杂,用java代码和上面的代码也差不多类似,如果直接用原生的api读取会稍微复杂,但如果我们使用Hive,Spark框架的时候,框架内部会自动帮我们完成压缩文件的读取或者写入,对用户透明,当然底层也是封装了不同压缩格式的读取和写入代码,这样以来使用者将会方便许多。



参考文章:

https://blog.matthewrathbone.com/2013/12/28/reading-data-from-hdfs-even-if-it-is-compressed

有什么问题可以扫码关注微信公众号:我是攻城师(woshigcs),在后台留言咨询。
技术债不能欠,健康债更不能欠, 求道之路,与君同行。

0
0
分享到:
评论

相关推荐

Global site tag (gtag.js) - Google Analytics