1 import org.apache.hadoop.conf.Configuration;
2 import org.apache.hadoop.fs.FileUtil;
3 import org.apache.hadoop.fs.Path;
4 import org.apache.hadoop.io.BytesWritable;
5 import org.apache.hadoop.io.SequenceFile;
6 import org.apache.hadoop.io.SequenceFile.Reader;
7 import org.apache.hadoop.io.SequenceFile.Writer;
8 import org.apache.hadoop.io.Text;
9 import org.slf4j.Logger;
10 import org.slf4j.LoggerFactory;
11
12 import javax.imageio.stream.FileImageOutputStream;
13 import java.io.File;
14 import java.io.FileInputStream;
15 import java.nio.charset.StandardCharsets;
16 import java.util.ArrayList;
17 import java.util.List;
18
19 public class MergeSmallFilesToSequenceFile {
20 private static Logger logger = LoggerFactory.getLogger(MergeSmallFilesToSequenceFile.class);
21 private static Configuration configuration = new Configuration();
22 private static List<String> smallFilePaths = new ArrayList<String>();
23
24 /**
25 *添加路径，读取文件夹下所有的文件绝对路径
26 * @param inputPath
27 * @throws Exception
28 */
29 public void addInputPath(String inputPath) throws Exception{
30
31 File file = new File(inputPath);
32
33 if(file.isDirectory()){
34 File[] files = FileUtil.listFiles(file);
35 for(File sFile:files){
36 smallFilePaths.add(sFile.getPath());
37 logger.info("添加小文件路径：" + sFile.getPath());
38 }
39 }else{
40 smallFilePaths.add(file.getPath());
41 logger.info("添加小文件路径：" + file.getPath());
42 }
43 }
44
45 /**
46 *合并小文件序列化存储到hdfs
47 * @throws Exception
48 */
49 public void mergeFile() throws Exception{
50
51 Writer.Option bigFile = Writer.file(new Path("/SequenceFile_Test/test/wangxin.test"));
52
53 Writer.Option keyClass = Writer.keyClass(Text.class);
54 Writer.Option valueClass = Writer.valueClass(BytesWritable.class);
55
56 Writer writer = SequenceFile.createWriter(configuration, bigFile, keyClass, valueClass);
57
58 Text key = new Text();
59 for(String path:smallFilePaths){
60 File file = new File(path);
61 long fileSize = file.length();
62 byte[] fileContent = new byte[(int)fileSize];
63 FileInputStream inputStream = new FileInputStream(file);
64 inputStream.read(fileContent, 0, (int)fileSize);
65
66 //String md5Str = DigestUtils.md5Hex(fileContent);
67
68 //logger.info("merge小文件："+path+",md5:"+md5Str);
69 key.set(path);
70
71 writer.append(key, new BytesWritable(fileContent));
72 }
73 writer.hflush();
74
75 /*for (String path : smallFilePaths) {
76 File file = new File(path);
77 file.deleteOnExit();
78 }*/
79 writer.close();
80 }
81
82 /**
83 * 在hdfs指定位置读取文件，读取k，v
84 * @throws Exception
85 */
86 public void readMergedFile() throws Exception{
87 MergeSmallFilesToSequenceFile smf = new MergeSmallFilesToSequenceFile();
88 configuration.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
89 Reader.Option file = Reader.file(new Path("hdfs://集群名/hdfs路径"));
90 Reader reader = new Reader(configuration, file);
91 Text key = new Text();
92 BytesWritable value = new BytesWritable();
93 while(reader.next(key, value)){
94 byte[] bytes = value.copyBytes();
95 //String md5 = DigestUtils.md5Hex(bytes);
96 String content = new String(bytes, StandardCharsets.UTF_8);
97 //logger.info("读取到文件："+key+",md5:"+md5+",content:"+content);
98 byte[] keyBytes = key.copyBytes();
99 String keyStr = new String(keyBytes, StandardCharsets.UTF_8);
100 //判断是不是图片，是图片转存到本地路径
101 if (keyStr.contains(".png")) {
102 String[] split = keyStr.split("\\\\");
103 String fileName = split[split.length - 1];
104 //拼接转存位置
105 smf.byte2image(bytes,"/linux路径" + fileName);
106 }else {
107 System.out.println("读取到文件：" + keyStr + ",content:" + content);
108 }
109
110 }
111 }
112
113 /**
114 *将byte数组转成图片
115 * @param data
116 * @param path
117 */
118 public void byte2image(byte[] data,String path){
119 if(data.length<3||path.equals("")) return;
120 try{
121 FileImageOutputStream imageOutput = new FileImageOutputStream(new File(path));
122 imageOutput.write(data, 0, data.length);
123 imageOutput.close();
124 System.out.println("转换图片成功 " + path);
125 } catch(Exception ex) {
126 System.out.println("Exception: " + ex);
127 ex.printStackTrace();
128 }
129 }
130
131
132 //测试
133 public static void main(String[] args) throws Exception {
134 MergeSmallFilesToSequenceFile msf = new MergeSmallFilesToSequenceFile();
135
136 /*List<String> smallFilePaths = new ArrayList<String>();
137 smallFilePaths = msf.getFiles("/tmp/logs",smallFilePaths);*/
138
139 /* msf.addInputPath("C:\\Users\\HR\\Desktop\\bigfile");
140
141 for (String smallFilePath : smallFilePaths) {
142 System.out.println(smallFilePath);
143 }
144
145 msf.mergeFile();*/
146
147 msf.readMergedFile();
148
149 /*msf.mergeFile();
150
151 msf.readMergedFile();*/
152 }
153 }
原文地址：https://www.cnblogs.com/China2008512/p/15428461.html
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。