一、结构化数据直接上传
如果我们拿到要上传的数据是结构化的，那么就不需要在对数据做处理， 直接从本地上传到HDFS上即可。
代码层面也比较简单：
public class UploadFileToHDFS {
public static void main(String[] args) throws Exception {
Configuration cfg = new Configuration();
FileSystem fs = FileSystem.get(new URI("hdfs://192.168.145.200:9000"),cfg);
// 自己修改登录hdfs的用户 也可用本地系统用户 去hdfs上授权
System.setProperty("HADOOP_USER_NAME","root");
/*
文件上传
*/
// 获得要上传文件的本地路径
Path src = new Path("f:/mydata/logs/Log_20200101.log");
// 上传的路径
Path dst = new Path("/tmp");
// 上传命令
fs.copyFromLocalFile(src,dst);
// 释放资源
fs.close();
}
}
二、半结构化 / 非结构化数据转化后再上传
有时我们拿到要上传的数据不一定是结构化的，可能是半结构化（JSON等）或者非结构化，这样即便上传也没什么意义。因此，我们先用Java语言，将其结构化了再上传。
例如：
我们拿到的数据：
我们上传后的数据;
package com.zyp.myhdfs.services;
import com.alibaba.fastjson.JSON;
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 java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/*
*
* todo：多线程传递多个文件到HDFS上
*
* */
public class UploadFileToHDFS2 {
/*
* 传进来路径 path 和文件名 filename， 将所有文件全部上传
*
* */
public void writeFileToHDFS(String path,String filename) {
FileSystem fs = null;
FileReader fis = null; // 输入流 读取本地文件
BufferedReader bis = null; // 对输入流整行整行读入 借用BufferedReader
FSDataOutputStream fos = null; // 输出流 往HDFS上输出数据
try {
fs = FileSystem.get(new URI("hdfs://192.168.145.200:9000"),new Configuration());
fis = new FileReader(path+"/"+filename);
bis = new BufferedReader(fis);
// 自己修改登录hdfs的用户 也可用本地系统用户 去hdfs上授权即可
System.setProperty("HADOOP_USER_NAME","root");
/*
用上面创建好的对象，往上写数据
*/
// 先去创建一个文件，然后往里面写
fos = fs.create(new Path("/logs/"+filename));
String line ="";
while ((line = bis.readLine()) !=null ){
Info info = JSON.parseObject(line, Info.class);
// System.out.println(info.getGoodid()+","+info.getMachine().getMemory());
// 利用Sting模板
String ctx = String.format("%s,%s,%s,%s,%s,%s,%s,%s,%s,%s\n", //设定好格式，通过映射去数据源中获取对应的数据数据信息
info.getMachine().getCpuType(),
info.getMachine().getMemory(),
info.getMachine().getCpuSeed(),
info.getActTime(),
info.getActType(),
info.getGoodId(),
info.getPage(),
info.getUseId(),
info.getBrowse().getBrowseType(),
info.getBrowse().getBrowseVersion());
fos.write(ctx.getBytes());
}
fos.flush();
} catch (IOException e) {
e.printStackTrace();
} catch (URISyntaxException e) {
e.printStackTrace();
} finally {
try {
fos.close();
bis.close();
fis.close();
// fs.close(); 这里关闭通道会影响其他传输
} catch (IOException e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) {
ExecutorService es = Executors.newFixedThreadPool(30); // 固定线程池
final UploadFileToHDFS2 ufh = new UploadFileToHDFS2();
final String filePath = "f:/mydata/logs";
// 循环获取所有的子文件
File file = new File(filePath);
String[] fs = file.list();
for (String fileName:fs){
es.execute(new Runnable() {
@Override
public void run() {
ufh.writeFileToHDFS(filePath,fileName);
}
});
}
es.shutdown();
}
}
三、分组整合文件后在上传至一个文件里
通过对上面在HDFS上的执行效果观察发现，本地一个文件可能很小，远远不够一个块大小（128M），但是他也单独占据一个block块。这样就造成了很大的资源浪费，这里考虑，将文件整合后再上传，尽可能的节省资源。
例如：
将一年内的日志信息，按月进行分类，HDFS上，每个月一个文件夹，然后往里面添加对应该月的数据信息
import com.alibaba.fastjson.JSON;
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 java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;
/*
* path: 源文件路径
* newFileName： hdfs上文件和文件夹的名字
* files： [log_20200101.log,log_20200102.log]
* todo: 考虑到每个文件占据一个block块，太浪费空间 这里将文件整合在一起存到HDFS上
* 按月份 将每个月的文件都写到一个文件夹下
* */
public class MyMergeFile {
public void batchWriteToHDFS(String path, String hdfsFileName, List<String> files) {
FSDataOutputStream fos = null;
try {
FileSystem fs = FileSystem.get(new URI("hdfs://192.168.145.200:9000"), new Configuration());
// 分割hdfs文件 获得文件夹名字
String folderName = hdfsFileName.split("_")[1];
// 自己修改登录hdfs的用户 也可用本地系统用户 去hdfs上授权即可
System.setProperty("HADOOP_USER_NAME", "root");
// 在hdfs下先创建文件夹
fs.mkdirs(new Path("/logs" + folderName));
// 在文件夹下创建一个文件
fos = fs.create(new Path("/logs" + folderName + "/" + hdfsFileName));
// 循环读取文件 并向hdfs文件中写入数据
for (String localFile : files) {
BufferedReader br = null;
try {
br = new BufferedReader(new FileReader(path + "/" + localFile));
String line = "";
while ((line = br.readLine()) != null) {
try {
Info info = JSON.parseObject(line, Info.class);
// System.out.println(info.getGoodid()+","+info.getMachine().getMemory());
// 利用Sting模板
String ctx = String.format("%s,%s,%s,%s,%s,%s,%s,%s,%s,%s\n",
info.getMachine().getCpuType(),
info.getMachine().getMemory(),
info.getMachine().getCpuSeed(),
info.getActTime(),
info.getActType(),
info.getGoodId(),
info.getPage(),
info.getUseId(),
info.getBrowse().getBrowseType(),
info.getBrowse().getBrowseVersion());
fos.write(ctx.getBytes());
} catch (Exception e) {
continue; // 防止源文件里JSON 格式错误 ，继续往下执行
}
}
fos.flush();
} catch (IOException e) {
e.printStackTrace();
} finally {
try {
br.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
} catch (IOException e) {
e.printStackTrace();
} catch (URISyntaxException e) {
e.printStackTrace();
} finally {
try {
fos.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
/*
* 将一年内的所有文件整合 分组 上传到hdfs
*
* */
public static void main(String[] args) {
/*
* 正常的文件分组
* */
// File file = new File("f:/mydata/logs");
//// 获取所有文件
// String[] files = file.list();
//// 按照月份分组 log_202001,log_202002
// Map<String, List<String>> map = new HashMap<>();
// for (String fn : files) {
// String keyName = fn.substring(0,10); // 截取出来所有文件的月份部分
// if (map.containsKey(keyName)){ // 判断该月份是否存在
// map.get(keyName).add(fn); // 存在该月份 就直接将文件存到该 key的list中
// }else {
// List<String> lst = new ArrayList<>(); // 不存在 就创建一个list 存放文件名
// lst.add(fn);
// map.put(keyName,lst);
// }
// }
// System.out.println(map);
/*
* 使用java工具 jdk必须是1.8以上
* */
File file = new File("f:/mydata/logs");
// 获取所有文件
String[] files = file.list();
List<String> lst = Arrays.asList(files);
// java 流式编程
Map<String, List<String>> mp = lst.stream().collect(
Collectors.groupingBy(line -> line.substring(0, 10))
);
// System.out.println(mp);
// 遍历hashmap
ExecutorService es = Executors.newFixedThreadPool(12);
final MyMergeFile mmf = new MyMergeFile();
for (String key : mp.keySet()) {
final String keyName = key;
final List<String> fs = mp.get(key);
es.execute(new Runnable() {
@Override
public void run() {
mmf.batchWriteToHDFS("f:/mydata/logs",keyName,fs);
}
});
}
es.shutdown();
}
}