Hadoop支持的文件格式之SequenceFile

电玩女神 2021-09-23 13:58 358阅读 0赞

### 文章目录 ###

*  0x00 文章内容
 *  0x01 SequenceFile格式概念
 *   *   *   *   *  1. SequenceFile是啥
 *  0x02 编码实现
 *   *   *   *   *  1. 写文件完整代码
                 *  2. 读文件完整代码
                 *  3. 写文件完整代码（HDFS）
                 *  4. 读文件完整代码（HDFS）
 *  0x03 校验结果
 *   *   *   *   *  1. 启动集群
                 *  2. 执行写SequenceFile文件格式代码
                 *  3. 执行读SequenceFile文件格式代码
                 *  4. 执行写SequenceFile文件格式代码（HDFS）
                 *  5. 执行读SequenceFile文件格式代码（HDFS）
 *  0xFF 总结

# 0x00 文章内容 #

Hadoop支持的四种常用的文件格式：`Text(csv)`、`Parquet`、`Avro`以及`SequenceFile`，非常关键！

# 0x01 SequenceFile格式概念 #

##### 1. SequenceFile是啥 #####

二进制格式。

# 0x02 编码实现 #

##### 1. 写文件完整代码 #####

package com.shaonaiyi.hadoop.filetype.sequence;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.FileSystem;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.SequenceFile;
    import org.apache.hadoop.io.Text;
    
    import java.io.IOException;
    import java.net.URI;
    
    /** * @Author shaonaiyi@163.com * @Date 2019/12/20 11:27 * @Description Hadoop支持的文件格式之写Sequence */
    public class SequenceFileWriter { 
    
        private static final String[] DATA = { 
                "shao, naiyi, bigdata, hadoop",
                "naiyi, bigdata, spark",
                "yi, two, a good man"
        };
        public static void main(String[] args) throws IOException { 
            String uri = "hdfs://master:9999/user/hadoop-sny/mr/filetype/sequence.seq";
    
            Configuration configuration = new Configuration();
            FileSystem fs = FileSystem.get(URI.create(uri), configuration);
            Path path = new Path(uri);
    
            IntWritable key = new IntWritable();
            Text value = new Text();
            SequenceFile.Writer writer = null;
            try { 
                writer = SequenceFile.createWriter(configuration,
                        SequenceFile.Writer.file(path), SequenceFile.Writer.keyClass(key.getClass()),
                        SequenceFile.Writer.valueClass(value.getClass()));
    
                for (int i = 0; i < 100; i++) { 
                    key.set(100 -i);
                    value.set(DATA[i % DATA.length]);
                    System.out.printf("[%s]\t%s\t%s\n", writer.getLength(), key, value);
                    writer.append(key, value);
                }
            } finally { 
                writer.close();
            }
    
        }
    
    }

代码解读：根据配置文件、文件路径、key类型、value类型此四个参数构建SequenceFile的Writer对象，然后循环append进key和value

##### 2. 读文件完整代码 #####

package com.shaonaiyi.hadoop.filetype.sequence;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.FileSystem;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.SequenceFile;
    import org.apache.hadoop.io.Writable;
    import org.apache.hadoop.util.ReflectionUtils;
    
    import java.io.IOException;
    import java.net.URI;
    /** * @Author shaonaiyi@163.com * @Date 2019/12/20 11:28 * @Description Hadoop支持的文件格式之读Sequence */
    public class SequenceFileReader { 
    
        public static void main(String[] args) throws IOException { 
            String uri = "hdfs://master:9999/user/hadoop-sny/mr/filetype/sequence.seq";
    
            Configuration configuration = new Configuration();
            FileSystem fs = FileSystem.get(URI.create(uri), configuration);
            Path path = new Path(uri);
    
            SequenceFile.Reader reader = null;
    
            try { 
                reader = new SequenceFile.Reader(configuration, SequenceFile.Reader.file(path));
                Writable key = (Writable)ReflectionUtils.newInstance(reader.getKeyClass(), configuration);
                Writable value = (Writable)ReflectionUtils.newInstance(reader.getValueClass(), configuration);
    
                long position = reader.getPosition();
    
                while (reader.next(key, value)) { 
                    String syncSeen = reader.syncSeen() ? "*" : "";
                    System.out.printf("[%s%s]\t%s\t%s\n", position, syncSeen, key, value);
                    position = reader.getPosition();
                }
    
            } finally { 
                reader.close();
            }
        }
    
    }

##### 3. 写文件完整代码（HDFS） #####

package com.shaonaiyi.hadoop.filetype.sequence;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.*;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
    import org.apache.hadoop.mapreduce.task.JobContextImpl;
    import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
    
    import java.io.IOException;
    
    /** * @Author shaonaiyi@163.com * @Date 2019/12/20 12:53 * @Description Hadoop支持的文件格式之写Sequence（HDFS） */
    public class MRSequenceFileWriter { 
    
        public static void main(String[] args) throws IOException, IllegalAccessException, InstantiationException, ClassNotFoundException, InterruptedException { 
            //1 构建一个job实例
            Configuration hadoopConf = new Configuration();
            Job job = Job.getInstance(hadoopConf);
    
            //2 设置job的相关属性
            job.setOutputKeyClass(LongWritable.class);
            job.setOutputValueClass(Text.class);
            job.setOutputFormatClass(SequenceFileOutputFormat.class);
    
            //3 设置输出路径
            FileOutputFormat.setOutputPath(job, new Path("hdfs://master:9999/user/hadoop-sny/mr/filetype/sequence"));
    
            //4 构建JobContext
            JobID jobID = new JobID("jobId", 123);
            JobContext jobContext = new JobContextImpl(job.getConfiguration(), jobID);
    
            //5 构建taskContext
            TaskAttemptID attemptId = new TaskAttemptID("attemptId", 123, TaskType.REDUCE, 0, 0);
            TaskAttemptContext hadoopAttemptContext = new TaskAttemptContextImpl(job.getConfiguration(), attemptId);
    
            //6 构建OutputFormat实例
            OutputFormat format = job.getOutputFormatClass().newInstance();
    
            //7 设置OutputCommitter
            OutputCommitter committer = format.getOutputCommitter(hadoopAttemptContext);
            committer.setupJob(jobContext);
            committer.setupTask(hadoopAttemptContext);
    
            //8 获取writer写数据，写完关闭writer
            RecordWriter<LongWritable, Text> writer = format.getRecordWriter(hadoopAttemptContext);
            String value = "shao";
            writer.write(new LongWritable(System.currentTimeMillis()), new Text(value));
            writer.close(hadoopAttemptContext);
    
            //9 committer提交job和task
            committer.commitTask(hadoopAttemptContext);
            committer.commitJob(jobContext);
        }
    
    }

##### 4. 读文件完整代码（HDFS） #####

package com.shaonaiyi.hadoop.filetype.sequence;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.*;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
    import org.apache.hadoop.mapreduce.task.JobContextImpl;
    import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
    
    import java.io.IOException;
    import java.util.List;
    import java.util.function.Consumer;
    /** * @Author shaonaiyi@163.com * @Date 2019/12/20 14:17 * @Description Hadoop支持的文件格式之读Sequence（HDFS） */
    public class MRSequenceFileReader { 
    
        public static void main(String[] args) throws IOException, IllegalAccessException, InstantiationException { 
            //1 构建一个job实例
            Configuration hadoopConf = new Configuration();
    
            Job job = Job.getInstance(hadoopConf);
    
            //2 设置需要读取的文件全路径
            FileInputFormat.setInputPaths(job, "hdfs://master:9999/user/hadoop-sny/mr/filetype/sequence");
    
            //3 获取读取文件的格式
            SequenceFileInputFormat inputFormat = SequenceFileInputFormat.class.newInstance();
    
            //4 获取需要读取文件的数据块的分区信息
            //4.1 获取文件被分成多少数据块了
            JobID jobID = new JobID("jobId", 123);
            JobContext jobContext = new JobContextImpl(job.getConfiguration(), jobID);
    
            List<InputSplit> inputSplits = inputFormat.getSplits(jobContext);
    
            //读取每一个数据块的数据
            inputSplits.forEach(new Consumer<InputSplit>() { 
                @Override
                public void accept(InputSplit inputSplit) { 
                    TaskAttemptID attemptId = new TaskAttemptID("jobTrackerId", 123, TaskType.MAP, 0, 0);
                    TaskAttemptContext hadoopAttemptContext = new TaskAttemptContextImpl(job.getConfiguration(), attemptId);
                    RecordReader<LongWritable, Text> reader = null;
                    try { 
                        reader = inputFormat.createRecordReader(inputSplit, hadoopAttemptContext);
                        reader.initialize(inputSplit, hadoopAttemptContext);
                        while (reader.nextKeyValue()) { 
                            System.out.println(reader.getCurrentKey());
                            System.out.println(reader.getCurrentValue());
                        }
                        reader.close();
                    } catch (IOException e) { 
                        e.printStackTrace();
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            });
    
        }
    
    }

# 0x03 校验结果 #

##### 1. 启动集群 #####

a. 启动HDFS集群，  
`start-dfs.sh`

##### 2. 执行写SequenceFile文件格式代码 #####

a. 直接在Win上执行，控制台会显示结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NoYW9jazIwMTg_size_16_color_FFFFFF_t_70]  
然后去集群也可以查看到结果：

hadoop fs -ls hdfs://master:9999/user/hadoop-sny/mr/filetype/
    hadoop fs -cat hdfs://master:9999/user/hadoop-sny/mr/filetype/sequence.seq

![在这里插入图片描述][2019122011422089.png]  
![在这里插入图片描述][20191220114312255.png]  
b. 其实，还可以通过如下命令以Text格式查看二进制文件

hadoop fs -text hdfs://master:9999/user/hadoop-sny/mr/filetype/sequence.seq

![在这里插入图片描述][20191220114352481.png]  
注意，此处控制台打印的日志与写进文件的内容不一样，所以看到控制台其实是多打印了`writer.getLength()`：

System.out.printf("[%s]\t%s\t%s\n", writer.getLength(), key, value);

PS：如果报权限错误：

Exception in thread "main" org.apache.hadoop.security.AccessControlException: Permission denied: user=Administrator, access=WRITE, inode="/user/hadoop-sny":hadoop-sny:supergroup:drwxr-xr-x

解决方案：需要去集群里修改权限

hadoop fs -mkdir -p hdfs://master:9999/user/hadoop-sny/mr/filetype
    hadoop fs -chmod 757 hdfs://master:9999/user/hadoop-sny/mr/filetype

##### 3. 执行读SequenceFile文件格式代码 #####

a. 也可以得到相应的结果  
![在这里插入图片描述][2019122011562055.png]

##### 4. 执行写SequenceFile文件格式代码（HDFS） #####

hadoop fs -ls hdfs://master:9999/user/hadoop-sny/mr/filetype/

![在这里插入图片描述][20191220141911748.png]  
![在这里插入图片描述][20191220142005816.png]

##### 5. 执行读SequenceFile文件格式代码（HDFS） #####

a. 可以看到代码里写进去的结果  
![在这里插入图片描述][20191220142233308.png]  
对应的打印代码为：

String value = "shao";
        writer.write(new LongWritable(System.currentTimeMillis()), new Text(value));

# 0xFF 总结 #

1.  Hadoop支持的文件格式系列：  
    [Hadoop支持的文件格式之Text][Hadoop_Text]  
    [Hadoop支持的文件格式之Avro][Hadoop_Avro]  
    [Hadoop支持的文件格式之Parquet][Hadoop_Parquet]  
    [Hadoop支持的文件格式之SequenceFile][Hadoop_SequenceFile]
2.  项目实战中，文章：[网站用户行为分析项目之会话切割（二）][Link 1]中使用的存储格式是`Parquet`。

--------------------

**福利：**  
[邵奈一的技术博客导航][Link 7]  
[邵奈一][CSDN] 原创不易，如转载请标明出处。

--------------------

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NoYW9jazIwMTg_size_16_color_FFFFFF_t_70]: /images/20210923/3f9cc7fc038a422895662dc014b7e5b2.png
[2019122011422089.png]: /images/20210923/7e8ab236993a4859b20aad66d072eb42.png
[20191220114312255.png]: /images/20210923/090cb0fcdf3440fe8593c20b49b6c950.png
[20191220114352481.png]: /images/20210923/ccf5cce7ca2246fd928181830868e4f4.png
[2019122011562055.png]: /images/20210923/9b0eac38cc204c7ebb82030b6a55ec82.png
[20191220141911748.png]: /images/20210923/93800a8e83934112b4f9aeba018cb4f5.png
[20191220142005816.png]: /images/20210923/1bc780f9eba640a0a64b5052c1c1f557.png
[20191220142233308.png]: /images/20210923/6aadee3e3e13406dabf630c494a257ec.png
[Hadoop_Text]: https://blog.csdn.net/shaock2018/article/details/103576005
[Hadoop_Avro]: https://blog.csdn.net/shaock2018/article/details/103581591
[Hadoop_Parquet]: https://blog.csdn.net/shaock2018/article/details/103592425
[Hadoop_SequenceFile]: https://blog.csdn.net/shaock2018/article/details/103629726
[Link 1]: https://blog.csdn.net/shaock2018/article/details/100775214
[Link 2]: https://blog.csdn.net/shaock2018/article/details/91184254
[Link 3]: https://img-blog.csdnimg.cn/20190608000417972.jpeg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NoYW9jazIwMTg=,size_16,color_FFFFFF,t_70
[Link 4]: https://img-blog.csdnimg.cn/20190514084602748.png
[Link 5]: https://weibo.com/5680168183/
[CSDN]: https://blog.csdn.net/shaock2018
[Link 6]: https://www.jianshu.com/u/6f033c6a3b90
[Link 7]: https://blog.csdn.net/shaock2018/article/details/89349249