package com.bjpowernode.hdfs.local;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
/**
 * ClassName:UploadLocalFileToHdfs <br/>
 * Function: 本地文件上传至hdfs. <br/>
 * Date:  2016年3月28日 下午10:06:05 <br/>
 * @author qiyongkang
 * @version 
 * @since JDK 1.6
 * @see  
 */
public class UploadLocalFileToHdfs {
 public static void main(String[] args) {
  Configuration conf = new Configuration();
  String localDir = "/home/qiyongkang";
  String hdfsDir = "/qiyongkang";
  try{
   Path localPath = new Path(localDir);
   Path hdfsPath = new Path(hdfsDir);
   FileSystem hdfs = FileSystem.get(conf);
   hdfs.copyFromLocalFile(localPath, hdfsPath);
  }catch(Exception e){
   e.printStackTrace();
  }
 }
}

package com.bjpowernode.hdfs.ftp;
import java.io.InputStream;
import org.apache.commons.net.ftp.FTP;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPFile;
import org.apache.commons.net.ftp.FTPReply;
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.IOUtils;
/**
 * ClassName:UploadFtpFileToHdfs <br/>
 * Function: TODO ADD FUNCTION. <br/>
 * Reason: TODO ADD REASON. <br/>
 * Date: 2016年3月28日 下午10:50:37 <br/>
 * 
 * @author qiyongkang
 * @version
 * @since JDK 1.6
 * @see
 */
public class UploadFtpFileToHdfs {
 public static void main(String[] args) {
  Configuration conf = new Configuration();
  loadFromFtpToHdfs("172.31.26.200", "qiyongkang", "qyk123456", "/www/input/", "/qiyongkang/", conf);
 }
 /**
  * 
  * loadFromFtpToHdfs:将数据从ftp上传到hdfs上. <br/>
  *
  * @author qiyongkang
  * @param ip
  * @param username
  * @param password
  * @param filePath
  * @param outputPath
  * @param conf
  * @return
  * @since JDK 1.6
  */
 private static boolean loadFromFtpToHdfs(String ip, String username, String password, String filePath,
   String outputPath, Configuration conf) {
  FTPClient ftp = new FTPClient();
  InputStream inputStream = null;
  FSDataOutputStream outputStream = null;
  boolean flag = true;
  try {
   ftp.connect(ip);
   ftp.login(username, password);
   ftp.setFileType(FTP.BINARY_FILE_TYPE);
   ftp.setControlEncoding("UTF-8");
   int reply = ftp.getReplyCode();
   if (!FTPReply.isPositiveCompletion(reply)) {
    ftp.disconnect();
   }
   FTPFile[] files = ftp.listFiles(filePath);
   FileSystem hdfs = FileSystem.get(conf);
   for (FTPFile file : files) {
    if (!(file.getName().equals(".") || file.getName().equals(".."))) {
     inputStream = ftp.retrieveFileStream(filePath + file.getName());
     outputStream = hdfs.create(new Path(outputPath + file.getName()));
     IOUtils.copyBytes(inputStream, outputStream, conf, false);
     if (inputStream != null) {
      inputStream.close();
      ftp.completePendingCommand();
     }
    }
   }
   ftp.disconnect();
  } catch (Exception e) {
   flag = false;
   e.printStackTrace();
  }
  return flag;
 }
}