我在 hdfs 的一个文件夹中有一堆 .gz 文件。我想将所有这些 .gz 文件解压缩到 hdfs 中的新文件夹中。我该怎么做？
【问题讨论】：
我可以考虑通过 3 种不同的方式来实现它。
使用 Linux 命令行
以下命令对我有用。
hadoop fs -cat /tmp/Links.txt.gz | gzip -d | hadoop fs -put - /tmp/unzipped/Links.txt
我的压缩文件是Links.txt.gz
输出存储在/tmp/unzipped/Links.txt
使用 Java 程序
在Hadoop The Definitve Guide 书中，有一个关于Codecs 的部分。在该部分中，有一个使用CompressionCodecFactory 解压缩输出的程序。我正在按原样重新生成该代码：
package com.myorg.hadooptests;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;
public class FileDecompressor {
public static void main(String[] args) throws Exception {
String uri = args[0];
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(URI.create(uri), conf);
Path inputPath = new Path(uri);
CompressionCodecFactory factory = new CompressionCodecFactory(conf);
CompressionCodec codec = factory.getCodec(inputPath);
if (codec == null) {
System.err.println("No codec found for " + uri);
System.exit(1);
}
String outputUri =
CompressionCodecFactory.removeSuffix(uri, codec.getDefaultExtension());
InputStream in = null;
OutputStream out = null;
try {
in = codec.createInputStream(fs.open(inputPath));
out = fs.create(new Path(outputUri));
IOUtils.copyBytes(in, out, conf);
} finally {
IOUtils.closeStream(in);
IOUtils.closeStream(out);
}
}
}
此代码将 gz 文件路径作为输入。
你可以这样执行：
FileDecompressor <gzipped file name>
例如当我为我的 gzip 文件执行时：
FileDecompressor /tmp/Links.txt.gz
我在以下位置获得解压缩文件：/tmp/Links.txt
它将解压缩的文件存储在同一文件夹中。所以你需要修改这段代码来接受2个输入参数：<input file path> and <output folder>。
一旦你让这个程序运行起来，你可以编写一个 Shell/Perl/Python 脚本来为你拥有的每个输入调用这个程序。
使用 Pig 脚本
您可以编写一个简单的 Pig 脚本来实现这一点。
我编写了以下脚本，它有效：
A = LOAD '/tmp/Links.txt.gz' USING PigStorage();
Store A into '/tmp/tmp_unzipped/' USING PigStorage();
mv /tmp/tmp_unzipped/part-m-00000 /tmp/unzipped/Links.txt
rm /tmp/tmp_unzipped/
当您运行此脚本时，解压缩的内容将存储在一个临时文件夹中：/tmp/tmp_unzipped。该文件夹将包含
/tmp/tmp_unzipped/_SUCCESS
/tmp/tmp_unzipped/part-m-00000
part-m-00000 包含解压缩的文件。
因此，我们需要使用以下命令显式重命名它，最后删除/tmp/tmp_unzipped 文件夹：
mv /tmp/tmp_unzipped/part-m-00000 /tmp/unzipped/Links.txt
rm /tmp/tmp_unzipped/
因此，如果您使用此 Pig 脚本，您只需要注意参数化文件名（Links.txt.gz 和 Links.txt）。
同样，一旦你让这个脚本工作，你可以编写一个 Shell/Perl/Python 脚本来为你拥有的每个输入调用这个 Pig 脚本。
【讨论】：
就我而言，我不想通过管道解压缩文件，因为我不确定它们的内容。相反，我想确保 zip 文件中的所有文件都将提取到 HDFS 上。
我创建了一个简单的 bash 脚本。评论应该让你知道发生了什么。下面有一个简短的描述。
#!/bin/bash
workdir=/tmp/unziphdfs/
cd $workdir
# get all zip files in a folder
zips=$(hadoop fs -ls /yourpath/*.zip | awk '{print $8}')
for hdfsfile in $zips
do
echo $hdfsfile
# copy to temp folder to unpack
hdfs dfs -copyToLocal $hdfsfile $workdir
hdfsdir=$(dirname "$hdfsfile")
zipname=$(basename "$hdfsfile")
# unpack locally and remove
unzip $zipname
rm -rf $zipname
# copy files back to hdfs
files=$(ls $workdir)
for file in $files; do
hdfs dfs -copyFromLocal $file $hdfsdir
rm -rf $file
done
# optionally remove the zip file from hdfs?
# hadoop fs -rm -skipTrash $hdfsfile
done
hdfs 目录中的所有*.zip 文件zip 复制到一个临时目录（在文件系统上）我设法让它使用/mypath/*/*.zip 为每个压缩文件中的许多 zip 文件使用子目录结构。
祝你好运:)
【讨论】：
如果您有压缩文本文件，hadoop fs -text 支持 gzip 以及其他常见的压缩格式（snappy、lzo）。
hadoop fs -text /tmp/a.gz | hadoop fs -put - /tmp/uncompressed_a
【讨论】：
您可以使用 hive 执行此操作（假设它是文本数据）。
create external table source (t str) location '<directory_with_gz_files>';
create external table target (t str) location '<target_dir>';
insert into table target select * from source;
数据将被解压缩成新的文件集。
如果您不想更改名称，并且您正在运行的节点上有足够的存储空间，则可以执行此操作。
hadoop fs -get <your_source_directory> <directory_name>
It will create a directory where you run hadoop command. cd to it and gunzip all the files
cd ..
hadoop fs -moveFromLocal <directory_name> <target_hdfs_path>
【讨论】：
STORED AS TEXTFILE 添加到源表定义中。
提供scala代码
import org.apache.hadoop.fs.{FSDataOutputStream, FileSystem, FileUtil, Path}
import org.apache.hadoop.io.compress.{CompressionCodecFactory, CompressionInputStream}
import org.apache.spark.sql.SparkSession
import org.apache.hadoop.io.IOUtils
val conf = new org.apache.hadoop.conf.Configuration()
def extractFile (sparkSession: SparkSession, compath : String, uncompPath :String): String = {
val fs = FileSystem.get(sparkSession.sparkContext.hadoopConfiguration)
val inputPath = new Path(compath)
val factory = new CompressionCodecFactory(sparkSession.sparkContext.hadoopConfiguration);
val codec = factory.getCodec(inputPath)
if (codec == null){
throw new RuntimeException(s"Not a valid codex $codec")
}
var in : CompressionInputStream = null;
var out : FSDataOutputStream = null;
try {
in = codec.createInputStream(fs.open(inputPath));
out = fs.create(new Path(uncompPath));
IOUtils.copyBytes(in, out, conf);
} finally {
IOUtils.closeStream(in);
IOUtils.closeStream(out);
}
uncompPath
}
【讨论】：