java hadoop hdfs 上写文件

项目中会用到往hdfs 上写文件  ，为下面kafka 往hdfs 上写文件做基础。

实例如下：

1、配置文件：com/xiefg/config/system.properties   

#以下是安装 hadoop 配置文件的路径

core.path=/opt/cloudera/parcels/CDH-5.5.0-1.cdh5.5.0.p0.8/lib/hadoop/etc/hadoop/core-site.xml
hdfs.path=/opt/cloudera/parcels/CDH-5.5.0-1.cdh5.5.0.p0.8/lib/hadoop/etc/hadoop/hdfs-site.xml
yarn.path=/opt/cloudera/parcels/CDH-5.5.0-1.cdh5.5.0.p0.8/lib/hadoop/etc/hadoop/yarn-site.xml
mapred.path=/opt/cloudera/parcels/CDH-5.5.0-1.cdh5.5.0.p0.8/lib/hadoop/etc/hadoop/mapred-site.xml

2、读取配置文件的工具类：
package com.xiefg.util;
import java.io.BufferedInputStream;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.PropertyResourceBundle;
import java.util.ResourceBundle;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;

/***
*
* @ClassName: PropertiesUtils
* @Description: TODO
* @author Comsys-xiefg
* @date 2017年2月5日 下午2:41:52
*
*/
public class PropertiesUtils {

private static ResourceBundle resources=null;

public static String HDFS_PATH = null;
public static String YARN_PATH = null;
public static String CORE_PATH = null;
public static String MAPRED_PATH = null;

static{
InputStream in;
try {
/*String config_path = System.getProperty("user.dir") + "/config/system.properties";
in = new BufferedInputStream(new FileInputStream(config_path));*/

in=Thread.currentThread().getContextClassLoader().getResourceAsStream(
"com/xiefg/config/system.properties");
resources = new PropertyResourceBundle(in);

//初始化hadoop配置
initHadoopConfig();

} catch (Exception e) {
e.printStackTrace();
}

}

/**
*
* @Title: initHadoopConfig
* @Description: 初始化hadoop 配置
* @return void 返回类型
* @throws
*/
public static void initHadoopConfig(){
HDFS_PATH = resources.getString("hdfs.path");
YARN_PATH = resources.getString("yarn.path");
CORE_PATH = resources.getString("core.path");
MAPRED_PATH = resources.getString("mapred.path");

}

public static Configuration getHDFSConf() {
Configuration conf = new Configuration();
conf.addResource(new Path(HDFS_PATH));
conf.addResource(new Path(CORE_PATH));
conf.addResource(new Path(MAPRED_PATH));
conf.addResource(new Path(YARN_PATH));
return conf;
}

/**
* 获取指定属性值
* @param property 属性名
* @return
*/
public static String getPropertiesValue(String property) {
String val = "";
try {
val = resources.getString(property);
} catch (Exception e) {
// ignore
e.printStackTrace();
}
return val;
}

public static void main(String[] args) {
Configuration conf = PropertiesUtils.getHDFSConf();
try {
HdfsFileUtil.appendFile(conf, "/test/kafka", "kafka test to hdfs xiefg".getBytes());
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}

//System.out.println(PropertiesUtils.getPropertiesValue(KafkaProperties.ZK));
}

}


3、hdfs 文件工具类
/**
*
*/
package com.xiefg.util;

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;

/**
*
* @ClassName: HdfsFileUtil
* @Description: TODO
* @author Comsys-xiefg
* @date 2017年2月6日 上午10:23:13
*
*/
public class HdfsFileUtil {
/***
*
* @Title: deleteHfile
* @Description: 删除hdfs指定目录的文件
* @param conf
* @param ioPath
* @throws IOException 设定文件
* @return void 返回类型
* @throws
*/
public static void deleteHfile(Configuration conf, String ioPath)
throws IOException {

FileSystem fileSystem=null;
try{
fileSystem = FileSystem.get(conf);
fileSystem.delete(new Path(ioPath), true);
}catch(Exception e){
e.printStackTrace();
}finally{
try {fileSystem.close();} catch (IOException e) {}
}
}

/**
*
* @Title: createFile
* @Description:写文件
* @param conf
* @param file
* @param content
* @throws IOException 设定文件
* @return void 返回类型
* @throws
*/

public static void createFile(Configuration conf,String file, String content) throws IOException {
FileSystem fs = FileSystem.get(conf);
byte[] buff = content.getBytes();
FSDataOutputStream os = null;
try {
os = fs.create(new Path(file));
os.write(buff, 0, buff.length);
os.flush();
System.out.println("Create: " + file);
} finally {
if (os != null)
os.close();
}
fs.close();
}
/**
*
* @Title: appendFile
* @Description: 创建文件并追加内容
* @param conf
* @param file
* @param buff
* @throws IOException 设定文件
* @return void 返回类型
* @throws
*/
public static void appendFile(Configuration conf,String file,byte[] buff) throws IOException {
FileSystem fs = FileSystem.get(conf);
Path path=new Path(file);
if (!fs.exists(path)) {
createFile(conf,file,new String(buff,"UTF-8"));
}else{
FSDataOutputStream os = null;
try {
os = fs.append(path);
os.write(buff, 0, buff.length);
os.flush();
//System.out.println("Create: " + file);
} finally {
if (os != null)
os.close();
fs.close();
}
}
}
/**
*
* @Title: isExist
* @Description: 判断文件是否存在
* @param path
* @param conf
* @return
* @throws IOException 设定文件
* @return boolean 返回类型
* @throws
*/

public static boolean isExist(String path,Configuration conf) throws IOException {
FileSystem fs=null;
Boolean isexists = null;
try {
Path p = new Path(path);
fs = p.getFileSystem(conf);
isexists=fs.exists(p);
}catch (Exception e) {
e.printStackTrace();
}finally{
fs.close();
}

return isexists;
}

public static void main(String[] args) throws Exception {

}
}


通过打jar 包  放在hadoop平台上 运行 命令：
hadoop jar /usr/etl.jar com.xiefg.util.PropertiesUtils

执行结果可以看到   hdfs 生成文件目录 和内容