Chanhae Oh

빅데이터, 분석

자바 인터페이스 읽기





하둡 URL로 데이터 읽기


하둡 파일시스템에서 파일을 읽는 가장 쉬운 방법은 java.net.URL 객체로 데이터를 읽는 스트림 하나를 여는 것


InputStream in = null;
try {
    in = new URL("hdfs://host/path").openStream();
}finally {
    IOUtils.closeStream(in);
}


자바가 하둡의 hdfs URL 스킴을 인식하기 위해서는 약간의 작업이 더 필요하며 FsURLStreamHandlerFactory의 인스턴스와 함께 URL 클래스의 setURLStreamHandlerFactory() 메소드를 호출하여 수행된다. 이 메소드는 JVM 하나 당 한 번씩만 호출할 수 있기 때문에 일반적으로 정적 블록에서 실행된다. 이러한 한계는 프로그램의 일부 다른 부분(아마도 제어할 수 있는 영역 밖에 있는 서드파티 컴포넌트)에서 URLStreamHandlerFactory를 설정하면 하둡 URL로부터 데이터를 읽을 수 없게 됨을 의미한다.

// 하둡 파일시스템의 파일을 URLStreamHandler를 사용하여 표준 출력으로 보여주기
public class URLCat {
    static {
        uRL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory());
    }

    public static void main(String[] args) throws Exception{
        InputStream in = null;
        try{
            in = new URL(args[0]).openStream();
            IOUtils.copyBytes(in, System.out, 4096, false);
        }finally {
            IOUtils.closeStream(in);
        }
    }
}

finally 구문에서 스트림을 닫고 입력 스트림에서 출력 스트림으로 바이트를 복하기 위해 하둡 클래스와 함께 IOUtils 클래스를 사용했다. copyBytes() 메소의 세 번째 인자는 버퍼 크기를, 네 번째 인자는 복사 완료 시 스트림의 닫기 여부를 의미한다. 입력 스트림은 직접 닫지만 System.out은 닫을 필요가 없다.

# 프로그램 실행 결과
export HADOOP_CLASSPATH=hadoop-examples.jar
hadoop URLCat hdfs://localhost/user/tom/quangle.txt

...


파일시스템 API로 데이터 읽기

URLStreamHandlerFactory 설정이 불가능한 경우 파일에 대한 입력 스트림을 열기 위해 FileSystem API를 사용해야 한다.


하둡 파일시스템의 파일은 하둡 Path 객체(의미상 로컬 파일시스템에 너무 고착되어버린 java.io.File 객체가 아니라)로 표현된다. 여기서 Path는 hdfs://localhost/user/tom/quangle/txt 같은 하둡 파일시스템 URL을 의미한다.

FileSystem은 일반적인 파일시스템 API로, 첫 단계는 접근할 파일시스템에 대한 인스턴스를 얻는 것이다. FileSystem 인스턴스를 얻을 수 있는 정적 팩토리 메소드는 다음과 같다.

public static FileSystem get(Configuration conf) throws IOExeption
public static FileSystem get(URI uri, Configuration conf) throws IOException
public static FileSystem get(URI uri, Configuration conf, String user) throws IOException

configuration 객체는 클라이언트나 서버의 환경 설정을 포함하고 있으며, 클래스패스에 있는 etc/hadoop/core-site.xml과 같은 설정 파일에서 관련 설정을 읽어 들인다. 첫 번째 메소드는 기본 파일시스템(core-site.xml 파일에 설정하며, 설정하지 않으면 기본 로컬 파일시스템이 사용됨)을 반환한다. 두 번째 메소드는 주어진 URI 스킴과 권한으로 파일시스템을 결정하며, URI에 스킴을 며시하지 않으면 기본 파일시스템으로 간주한다. 세 번째 메소드는 특정 사용자를 명시하여 파일시스템을 추출하며, 보안 차원에서 중요하다.


로컬 파일시스템 인스턴스만 얻을 때는 getLocal() 메소드를 사용하는 것이 편리하다.

public static LocalFileSystem getLocal(Configuration conf) throws IOException


FileSystem 인스턴스를 얻으면 open() 메소드를 호출하여 파일에 대한 입력 스트림을 열 수 있다.

public FSDataInputStream open(Path f) throws IOException
public abstract FSDataInputStream open(Path f, int bufferSize) throws IOException

첫 번째 메소드는 4KB의 기본 버퍼 크기를 사용.

// FileSystem API를 직접 사용하여 하둡 파일시스템의 파일을 표준 출력으로 보여주기
public class FileSystemCat {
    public static void main(String[] args) throws Exception {
        String uri = args[0];
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(URI.create(uri), conf);
        InputStream in = null;
        try{
            in = fs.open(new Path(uri));
            IOUtils.copyBytes(in, System.out, 4096, false);
        }finally {
            IOUtils.closeStream(in);
        }
    }
}

# 프로그램 실행 결과
hadoop FileSystemCat hdfs://localhost/user/tom/quangle.txt

...


FSDataInputStream

FileSystem의 open() 메소드는 표준 java.io 클래스가 아닌 FSDataInputStream 클래스를 반환한다. 이 클래스는 랜덤 접근을 지원하기 위해 java.io.DataInputStream을 특별히 변경한 것이며, 스트림의 어떤 부분이든 읽을 수 있다.

package org.apache.hadoop.fs;

public class FSDataInputStream extends DataInputStream implements Seekable, PositionedReadable {
    ...
}

Seekable 인터페이스는 파일에서 특정 위치로 이동하는 것을 허용하며, 파일의 시작점에서 현재 오프셋을 얻는 getPos()와 같은 쿼리 메소드를 제공한다.

public interface Seekable {
    void seek(long pos) throws IOException;
    long getPos() throws IOExcption;
}

파일의 길이보다 더 큰 값으로 seek() 메소드를 호출하면 IOException을 일으킨다. 현재 위치에서 뒤로 스트림의 위치를 이동시키는 java.io.InputStream 클래스의 skip() 메소드와 달리 seek() 메소드는 파일에서 임의 위치나 절대 위치로 이동할 수 있다.

// seek를 사용해서 하둡 파일시스템의 파일을 표준 출력으로 두 번 보여주기
public class FileSystemDoubleCat {
    public static void main(String[] args) throws Exception {
        String uri = args[0];
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(URI.create(uri), conf);
        FSDataInputStream in = null;
        try{
            in = fs.open(new Path(uri));
            IOUtils.copyBytes(in, System.out, 4096, false);
            in.seek(0);
            IOUtils.copyBytes(in, System.out, 4096, false);
        }finally {
            IOUtils.closeStream(in);
        }
    }
}

hadoop FileSystemDoubleCat hdfs://localhost/user/tom/quangle.txt

...
...

FSDataInputSTream 역시 주어진 오프셋에서 파일의 일부를 읽기 위한 PositionedReadable 인터페이스를 제공한다.

public interface PositionedReadable{
    public int read(long position, byte[] buffer, int offset, int length) throws IOException;

    public void readFully(long position, byte[] buffer, int offset, int length) throws IOException;

    public void readFully(long postion, byte[] buffer) throws IOException;
}

read() 메소드는 파일의 주어진 position에서 length 만큼의 바이트를 읽어서 buffer의 주어진 offset에 그 내용을 복사한다. 반환값은 실제로 읽은 바이트 크기다. 호출자는 반환값이 length 보다 작은지 반드시 확인해야 한다. EOFExtption이 발생하는 파일의 끝이 아니라면 readFully() 메소드는 length 만큼의 바이트(또는 바이트 배열 buffer의 단순한 buffer.length 바이트)를 버퍼로 읽을 것이다.


이러한 모든 메소드는 파일의 현재 오프셋을 유지하고 스레드 안전을 지원한다(동시 접속을 고려하여 FSDataINputStream을 설계하지 않았기 때문에 다중 인스턴스를 생성하는 방법을 권장한다). 따라서 파일 전체를 읽는 중이더라도 메타데이터와 같은 파일의 다른 부분에 쉽게 접근할 수 있다.


seek() 메소드를 호출하는 것은 상대적으로 비용이 많이 드는 연산이므로 꼭 필요할 대만 사용해야 한다. 어플리케이션이 다수의 seek()를 수행하는 방식이 아닌 맵리듀스와 같은 스트리밍 데이터 기반의 접근 패턴을 사용하도록 구조화하라.