package hadoop;

import java.io.BufferedReader;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.URL;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.fs.FsUrlStreamHandlerFactory;

public class Hellohdfs {
    
    private static final String PATH="hdfs://192.168.199.132:9000/test.txt";

    public static void main(String[] args) throws Exception {
        //设定开启HDFS协议
        URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory());

        //开启URL
        InputStream in =  new URL(PATH).openStream();

        //读取文件
        BufferedReader reader = new BufferedReader(new InputStreamReader(in));
        String line = null;
        while((line =reader.readLine())!= null){
            System.out.println(line.toString());
        }

       
//      org.apache.hadoop.io.IOUtils.copyBytes(in, System.out, 1024,true);
    }
}

 
此时会出现connected refused问题这时候按以下步骤修改
 
把core-site.xml中的localhost 改为本机的ip地址
 
然后修改/etc/hosts文件
 
把localhost改为自己的ip地址
 
并添加
 
0.0.0.0  192.168.199.132（自己的ip地址）

项目中需要把消息写到hdfs文件系统中去,所以看了一下如何通过java来操作hdfs文件,这里记录一下.
不过效果不是很满意.需要优化的地方还很多,希望博客大神能指导一下.


下面是基本操作工具类
package testHbase2.common.util;

import java.io.IOException;
import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

/**
 * 操作hdfs文件工具
 * 
 * @author yhy
 *
 */
public class HDFSUtils {
  public static void appendHDFS(String hdfs, String data) {
    Configuration conf = new Configuration();
    conf.setBoolean("dfs.support.append", true);//有说法是添加了这个属性,就能再hdfs问价后面追加,但没有实现.
    FileSystem fs;
    try {
      fs = FileSystem.get(URI.create(hdfs), conf);
      Path path = new Path(hdfs);
      if(!fs.exists(path)){//判断文件是否存在
        FSDataOutputStream out = fs.create(path);
        out.write(data.getBytes("UTF-8"));
        out.close();
      }else{//看网上说了很多hdfs文件最好不要追加操作,能理解缘由,但没有想到更好的办法来实现业务需求,所以用了一个很笨的办法.就是先读取,再写入,但好像有点问题.
        FSDataInputStream in = fs.open(path);
        byte[] ioBuffer = new byte[10240];
        int readLen = in.read(ioBuffer);
        while (readLen != -1) {
          readLen = in.read(ioBuffer);
        }
        in.close();
        String str = new String(ioBuffer) + "\r\n"+data;
        FSDataOutputStream out = fs.create(path);
        out.write(str.getBytes("UTF-8"));
        out.close();
      }
      fs.close();
    } catch (IOException e) {
      e.printStackTrace();
    }
  }
}


下面是调用main方法
package testHbase2.controller;

import testHbase2.common.util.HDFSUtils;



public class WsController {

  public static void main(String[] args) {//这里的9000端口是可以在hadoop安装环境下的配置文件hdfs-site.xml中配置的.(好像是这个文件.不是的话百度一下就出来了,很简单)
    HDFSUtils.appendHDFS("hdfs://192.168.32.128:9000/usr/local/hadoop/t.txt", "测试数据222");
  }
  
}






 

 1.java.io.IOException:No FileSystem for scheme:hdfs

 解决：在pom.xml中漏添加了依赖 。下面3个依赖必不可少 

 <dependency>

 <groupId>org.apache.hadoop</groupId>

 <artifactId>hadoop-common</artifactId>

 <version>2.7.1</version>

 </dependency>

 

 <dependency>

 <groupId>org.apache.hadoop</groupId>

 <artifactId>hadoop-hdfs</artifactId>

 <version>2.7.1</version>

 </dependency>

 <dependency>

 <groupId>org.apache.hadoop</groupId>

 <artifactId>hadoop-client</artifactId>

 </dependency>

 

 

 2.Checksum error: file:/F:/test.txt at 0 exp: 1871533573 got: 2052517348

 解决：通过java程序将文件上传到hdfs上时，hadoop的api会通过fs.FSInputChecker判断需要上传的文件是否存在.crc校验文件。

 如果存在.crc校验文件，则会进行校验。如果校验失败，则不会上传该文件。.crc校验文件和需要上传的文件在同一目录下。

 解决办法为删除该.crc文件即可。

 

 3.Call From kingdee-PC/192.168.56.1 to 172.20.176.134:8020 failed on connection exception: java.net 

 .ConnectException: Connection refused: no further information; For more details see: 

 http://wiki.apache.org/hadoop/ConnectionRefused 

 解决：打开hadoop的安装目录hadoop-2.7.3/etc/hadoop下的文件core-site.xml，看到如下内容：

 <property>

 <name>fs.defaultFS</name>

 <value>hdfs://主机名:9000</value>

 </property>

 因此需要在代码中hdfs的uri中添加对应的端口号9000. 
 
 

使用Java API操作HDFS文件系统
 
关键点：
 
1）创建Configuration
2）获取FileSystem
3）…就是你的HDFS API的操作
 
首先创建FileSystem
 
    构造一个访问指定HDFS系统的客户端对象
    第一个参数：HDFS的URI
    第二个参数：客户端指定的配置参数
    第三个参数：客户端的身份，说白了就是用户名
 
 
 
FileSystemfileSystem = FileSystem.get(new URI(“hdfs://hadoop000:8020”), new configuration, “hadoop”);
 
 
1.创建HDFS文件夹
 
@Test
    public void mkdir() throws Exception {
        fileSystem.mkdirs(new Path("/hdfsapi/test"));
    }
 
2.查看HDFS内容
 
import org.apache.hadoop.io.IOUtils;
 
@Test
    public void text()throws Exception {
        FSDataInputStream in = fileSystem.open(new Path("/cdh_version.properties"));
        IOUtils.copyBytes(in, System.out, 1024);
    }
 
3.创建文件
 
 public void create()throws Exception {
	        FSDataOutputStream out = fileSystem.create(new Path("/hdfsapi/test/b.txt"));
	        out.writeUTF("hello DM replication 1");
	        out.flush();
	        out.close();
    }
 
4.测试文件名更改
 
 @Test
    public void rename() throws Exception {
        Path oldPath = new Path("/hdfsapi/test/b.txt");
        Path newPath = new Path("/hdfsapi/test/c.txt");
        boolean result = fileSystem.rename(oldPath, newPath);
        System.out.println(result);

    }
 
5.拷贝本地文件到HDFS文件系统(put)
 
@Test
    public void copyFromLocalFile() throws Exception {
        Path src = new Path("/Users/rocky/data/hello.txt");
        Path dst = new Path("/hdfsapi/test/");
        fileSystem.copyFromLocalFile(src,dst);
    }
 
6.拷贝HDFS文件到本地(get)
 
@Test
    public void copyToLocalFile() throws Exception {
        Path src = new Path("/hdfsapi/test/hello.txt");
        Path dst = new Path("/Users/rocky/tmp/software");
        fileSystem.copyToLocalFile(src, dst);
    }
 
7.查看目标文件夹下的所有文件
 
@Test
    public void listFiles() throws Exception {
        FileStatus[] statuses = fileSystem.listStatus(new Path("/hdfsapi/test"));

        for(FileStatus file : statuses) {
            String isDir = file.isDirectory() ? "文件夹" : "文件";
            String permission = file.getPermission().toString();
            short replication = file.getReplication();
            long length = file.getLen();
            String path = file.getPath().toString();


            System.out.println(isDir + "\t" + permission
                    + "\t" + replication + "\t" + length
                    + "\t" + path
            );
        }

    }
 
8.递归查看目标文件夹下的所有文件
 
@Test
    public void listFilesRecursive() throws Exception {

        RemoteIterator<LocatedFileStatus> files = fileSystem.listFiles(new Path("/hdfsapi/test"), true);

        while (files.hasNext()) {
            LocatedFileStatus file = files.next();
            String isDir = file.isDirectory() ? "文件夹" : "文件";
            String permission = file.getPermission().toString();
            short replication = file.getReplication();
            long length = file.getLen();
            String path = file.getPath().toString();


            System.out.println(isDir + "\t" + permission
                    + "\t" + replication + "\t" + length
                    + "\t" + path
            );
        }
    }
 
9.查看文件块信息
 
@Test
    public void getFileBlockLocations() throws Exception {

        FileStatus fileStatus = fileSystem.getFileStatus(new Path("/hdfsapi/test/jdk.tgz"));
        BlockLocation[] blocks = fileSystem.getFileBlockLocations(fileStatus,0,fileStatus.getLen());

        for(BlockLocation block : blocks) {

            for(String name: block.getNames()) {
                System.out.println(name +" : " + block.getOffset() + " : " + block.getLength() + " : " + block.getHosts());
            }
        }
    }
 
10删除文件
 
@Test
    public void delete() throws Exception {
        boolean result = fileSystem.delete(new Path("/hdfsapi/test/jdk.tgz"), true);
        System.out.println(result);
    }