在hadoop中下载序列文件

zqry0prt  于 2021-06-03  发布在  Hadoop
关注(0)|答案(2)|浏览(349)

我无法将二进制文件(在hadoop中存储为序列文件)复制到本地计算机。问题是,我从hdfs下载的二进制文件不是我运行map reduce任务时生成的原始二进制文件。我在google上搜索了类似的问题,我想问题是当我把序列文件复制到本地机器上时,我得到了序列文件的头加上原始文件。
我的问题是:有没有办法避免下载头文件,但仍然保留我原来的二进制文件?
我可以考虑两种方法:
我可以将二进制文件转换成其他格式,比如文本,这样就可以避免使用sequencefile。在我复制本地文件之后,我把它转换回二进制文件。
我仍然使用序列文件。但是当我生成二进制文件时,我也会生成一些关于相应序列文件的元信息(例如头的长度和文件的原始长度)。在我完成copytolocal之后,我使用下载的二进制文件(包含头文件等)以及元信息来恢复我原来的二进制文件。
我不知道哪一个可行。有人能给我一个解决办法吗?你能给我看一下你给出的解决方案的一些示例代码吗?
我非常感谢你的帮助。

blmhpbnm

blmhpbnm1#

我找到了解决这个问题的方法。因为下载序列文件会给你在二进制文件中的头和其他魔术字,我避免这个问题的方法是将我的原始二进制文件转换成base64字符串,并将其作为文本存储在hdfs中,当下载编码的二进制文件时,我将其解码回我的原始二进制文件。
我知道这将需要额外的时间,但目前我没有找到任何其他解决这个问题的办法。直接删除序列文件中的头和其他魔术字的困难之处在于hadoop可能会在我的二进制文件之间插入一些单词“sync”。
如果有人对这个问题有更好的解决办法,我很高兴听到这个消息

hyrbngr7

hyrbngr72#

使用mapreduce代码读取sequencefile,并使用sequencefileinputformat作为inputfileformat读取hdfs中的序列文件。这会将文件拆分为键值对,并且该值只包含可用于创建二进制文件的二进制文件内容。
下面是一个代码片段,用于拆分由多个图像组成的序列文件,并将其拆分为单独的二进制文件,然后将其写入本地文件系统。

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class CreateOrgFilesFromSeqFile {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        if (args.length !=2){
            System.out.println("Incorrect No of args (" + args.length + "). Expected 2 args: <seqFileInputPath> <outputPath>");
            System.exit(-1);
        }

        Path seqFileInputPath = new Path(args[0]);
        Path outputPath = new Path(args[1]);

        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "CreateSequenceFile");

        job.setJarByClass(M4A6C_CreateOrgFilesFromSeqFile.class);
        job.setMapperClass(CreateOrgFileFromSeqFileMapper.class);

        job.setInputFormatClass(SequenceFileInputFormat.class);

        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(Text.class);

        FileInputFormat.addInputPath(job, seqFileInputPath);
        FileOutputFormat.setOutputPath(job, outputPath);

        //Delete the existing output File
        outputPath.getFileSystem(conf).delete(outputPath, true);

        System.exit(job.waitForCompletion(true)? 0 : -1);

    }

}

class CreateOrgFileFromSeqFileMapper extends Mapper<Text, BytesWritable, NullWritable, Text>{

    @Override
    public void map(Text key, BytesWritable value, Context context) throws IOException, InterruptedException{

        Path outputPath = FileOutputFormat.getOutputPath(context);
        FileSystem fs = outputPath.getFileSystem(context.getConfiguration());

        String[] filePathWords = key.toString().split("/");
        String fileName = filePathWords[filePathWords.length-1];

        System.out.println("outputPath.toString()+ key: " + outputPath.toString() + "/" + fileName + "value length : " + value.getLength());

        try(FSDataOutputStream fdos = fs.create(new Path(outputPath.toString() + "/" + fileName)); ){

            fdos.write(value.getBytes(),0,value.getLength());
            fdos.flush();
        }

            //System.out.println("value: " + value + ";\t baos.toByteArray().length: " + baos.toByteArray().length);
            context.write(NullWritable.get(), new Text(outputPath.toString() + "/" + fileName));            
    }
}

相关问题