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

本文主要是介绍如何在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)).asScalaval 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)，在后台留言咨询。技术债不能欠，健康债更不能欠，求道之路，与君同行。

输入图片说明