빅데이터 - 하둡, 하이브로 시작하기 Hadoop, Hive

https://wikidocs.net/book/2203

1회 : 2012.09 | 아파치 플룸을 통한 로그 수집 

2회 : 2012.10 | 아파치 하둡 활용한 로그 저장 및 처리

3회 : 2012.11 | 아파치 하이브 로그 분석

http://hadoop.apache.org/

http://hive.apache.org/

아파치 하둡에는 그 유용성을 더 넓혀 주는 다양한 에코시스템이 있다.

하둡을 중심으로 플룸(Flume), 하이브(Hive), HBase, ZooKeeper, Oozie, Pig, Sqoop, whirr 등 다양한 에코시스템 생태계가 활성화 돼 있다.

* 빅데이터 3대 활용 요소

NoSQL(Not only Sql)은 지금까지 사용되왔던 관계형 데이터베이스 모델에 얽매이지 않고 비테이블 기반이다.

NoSQL은 추가/추출 Operation과 레코드 저장 기능에 대해서 최적화가 적용되어 대용량 데이터 처리에 대해서

기존 관계형 데이터베이스가 가지고 있던 단점을 보완할 수 있다.

* 클라우드 서비스에 적용되는 데이터베이스 솔루션

HBase는 구글 빅테이블의 클론 솔루션이며 무한한 데이터 수용 확장성을 지원한다.

HBase는 HDFS(Hadoop Distributed File System)에 구현한 분산 칼럼 기반 데이터베이스이며,

 대규모 데이터셋에 실시간으로 랜덤 엑세스가 필요할 때 사용할 수 있는 Hadoop 응용프로그램이다.

* Oracle NoSQL Database 특징

• 단순한 데이터 모델 지원(Major/Sub key를 사용한 key/value 쌍 지원)
• 단순한 프로그래밍 모델 지원(ACID transaction과 JSON 지원)
• Oracle DB & Hadoop 연동
• 확장가능한 throughput 제공
• 동적인 용량 추가 동작 지원
• 설정 가능한 다중 복제를 이용한 높은 가용성 제공

* MongoDB 특징

• 실행용 공식 Binary 파일은 Windows,Mac, Linux, Solaris에서 사용가능
• 공식 드라이버는 C,C#,C++, Haskell, JAVA, javascript, Perl, PHP, Python, Ruby, Scala에서 사용가능
• 임시(Ad-hoc) Javascript Query 지원(모든 문서 속성에서 기준을 사용해 데이터 검색)
• Query에서 정규 표현식 지원
• MongoDB의 Query 결과는 limit(),skip(),sort(),count(),distinct(),group()을 포함해 필터링과 수집 및 정렬에 필요한 다양한 함수를 제공하는 커서에 저장
• 고급 수집용 map/reduce 구현
• GridFS를 사용하는 대용량 파일 스토리지
• RDBMS 형태의 속성 인덱싱 지원
• Query 최적화 기능 지원
• MySQL 과 비슷한 Master/Slave 복제
• 참조쿼리를 허용하는 콜렉션 기반 오브젝트 스토리지
• Auto-sharding 을 이용한 수평적 확장
• 고성능의 동시성을 구현 가능 : 제자리 쓰기(In-place update)

클라우드 컴퓨팅 입문

Linux와 Apache Hadoop을 사용한 클라우드 컴퓨팅

클라우드 컴퓨팅 소개

최근에는 클라우드 컴퓨팅에 관한 선전이 증가하고 있으며 이러한 현상은 IT 산업에 대한 앞으로의 경향을 보여준다. 간단히 말해서 클라우드 컴퓨팅은 외부 환경에서 서비스 형태로 제공되는 확장 가능한 컴퓨팅 리소스를 사용량에 따라 요금을 지불하는 방식으로 사용하는 것이라고 정의할 수 있다. 인터넷을 통해 "클라우드"에 있는 모든 리소스를 액세스할 수 있으며 컴퓨팅 능력, 대역폭, 스토리지, 보안 및 신뢰성에 관해 걱정할 필요가 없다.

이 기사에서는 가상 Linux® 서버를 임대할 수 있는 Amazon EC2와 같은 클라우드 컴퓨팅을 간략하게 소개한 후 가상 Linux 서버에 구축하여 클라우드 컴퓨팅 프레임워크를 설정할 수 있는 오픈 소스 MapReduce 프레임워크인 Apache Hadoop을 소개한다. 그러나 벤더에서 호스트하는 VM에서만 Hadoop을 전개할 수 있는 것은 아니며 사용자는 실제 시스템에서 실행되는 일반 Linux OS에서 Hadoop을 전개할 수 있다.

Apache Hadoop을 자세히 살펴보기 전에 클라우드 컴퓨팅 시스템의 구조에 대해 간략히 소개한다. 그림 1에는 클라우드 컴퓨팅의 계층과 기존의 몇 가지 오퍼링이 표시되어 있다. 클라우드 컴퓨팅의 계층에 관한 자세한 사항은 참고자료 섹션을 참고한다.

IaaS(Infrastructure-as-a-Service)와 같은 인프라스트럭처는 인프라스트럭처(컴퓨팅 리소스 및 스토리지)를 서비스 형태로 임대하는 것을 말한다. IaaS에는 서비스 품질에 특정한 제한조건(특정 운영 체제 및 소프트웨어를 실행할 수 있는 기능)을 두고 컴퓨터나 가상 호스트 또는 데이터 센터를 임대할 수 있는 기능이 있다. Amazon EC2는 이러한 계층에서 IaaS로서 역할을 하여 사용자에게 가상 호스트를 제공한다. 이러한 플랫폼(Platform-as-a-Service 또는 PaaS)에서는 인프라스트럭처에서 "클라우드" 컴퓨팅에 API 기능을 제공하는 서비스나 소프트웨어 프레임워크에 집중한다. Apache Hadoop은 PaaS로서 역할을 하며 가상 서버에서 클라우드 컴퓨팅 플랫폼으로 구축된다.

Amazon EC2

Amazon EC2는 사용자가 가상 시스템을 다양한 기능(CUP, 디스크, 메모리 등)과 함께 요청할 수 있는 웹 서비스이다. 사용자는 컴퓨터를 사용한 시간에 대해서만 비용을 지불하며 호스팅 작업은 Amazon에서 처리한다.

이러한 인스턴스 즉, AMI(Amazon Machine Image)는 Linux를 기반으로 하며 원하는 모든 애플리케이션이나 소프트웨어를 실행할 수 있다. Amazon에서 서버를 임대한 후에야 일반 SSH 도구를 사용하여 연결을 설정하여 해당 서버를 실제 서버처럼 조작할 수 있다.

이 기사에서는 EC2를 자세히 소개하지 않는다. 추가 정보는 참고자료 섹션을 참고한다.

Hadoop 클라우드 컴퓨팅 프레임워크에 대한 베스트 프랙티스는 AMI에서 이 프레임워크를 전개하는 것이며 컴퓨팅 기능, 대역폭, 스토리지 등이 문제가 되지 않는다면 AMI을 통해 클라우드 기능을 활용할 수 있다. 그러나 Hadoop은 클라우드 솔루션에서만 전개되는 것이 아니므로 이 기사의 다음 부분에서는 로컬에서 호스트되는 Linux 서버의 VMware 이미지에서 Hadoop을 구축한다. 그전에 Apache Hadoop을 간단히 소개한다.

Apache Hadoop

Apache Hadoop은 대량의 데이터를 분산 처리할 수 있는 소프트웨어 프레임워크(플랫폼)이다. 2006년에 도입되었으며 Google, Yahoo! 및 IBM 등에서 지원한다. 이 Apache Hadoop을 PaaS 모델로 생각할 수도 있다.

디자인의 핵심은 MapReduce를 구현했다는 점과 MapReduce(Google의 자료에서 소개됨)와 Google File System에서 영향을 받은 HDFS(Hadoop Distributed File System)에 있다.

MapReduce

MapReduce는 Google에서 도입한 소프트웨어 프레임워크로 컴퓨터 클러스터나 노드 클러스터에서 대용량 데이터 세트 분산 컴퓨팅을 지원한다. MapReduce에는 Map 프로세스와 Reduce 프로세스가 결합되어 있다.

Map 프로세스에서는 마스터 노드가 작업을 입력 받아서 더 작은 하위 작업으로 나눈 후 작업자 노드에 분배한다.

작업자 노드는 이러한 하위 작업을 처리한 후 그 결과를 다시 마스터 노드에 전달한다.

그러면 Reduce 프로세스에서 마스터 노드가 모든 하위 작업의 결과를 받아서 서로 결합시켜 원래 작업의 결과인 출력을 얻는다.

MapReduce 플로우에 관한 개념은 그림 2를 참조한다.

MapReduce의 장점은 맵과 리덕션 조작을 분산 처리할 수 있다는 점이다. 각 맵핑 조작은 독립적이기 때문에 모든 맵은 병렬로 실행될 수 있으며 따라서 총 컴퓨팅 시간을 줄일 수 있다.

HDFS

HDFS에 대한 자세한 소개와 이 파일 시스템을 조작하는 방법은 이 기사에서 다루지 않는다. 추가 정보는 참고자료 섹션을 참조한다.

일반 사용자 관점에서 보면 HDFS는 기존 파일 시스템과 동일하다. 특정 디렉토리 경로를 사용하여 파일에 CRUD 조치를 실행할 수 있다. 그러나 분산 스토리지의 특성으로 인해 각각 역할이 다른 "NameNode"와 "DataNode"가 있다.

NameNode는 DataNode의 마스터이다. NameNode는 HDFS 내에서 메타데이터 서비스를 제공한다. 메타데이터는 DataNode의 파일 맵핑을 표시한다. 또한 조작 명령을 승인하고 어느 DataNode가 조작과 복제를 실행해야 하는지 결정한다.

DataNode는 HDFS의 스토리지 블록으로 동작한다. 또한 NameNode에서 수신된 블록을 작성하고, 삭제하거나 복제하는 명령에 응답한다.

JobTracker 및 TaskTracker

애플리케이션을 실행할 때에는 HDFS에 있는 입력과 출력 디렉토리를 함께 제공해야 한다. JobTracker는 MapReduce 애플리케이션을 실행하기 위한 단일한 제어점으로, 작성될 TaskTracker와 종속 작업의 수를 결정하여 각 하위 작업을 TaskTracker에 할당한다. 각 TaskTracker가 상태를 보고하면 완료된 작업은 JobTracker로 돌아간다.

일반적으로 하나의 마스터 노드가 NameNode와 JobTracker로 동작하며 슬레이브 노드는 DataNode와 TaskTracker로 동작한다. Hadoop Cluster에 대한 개념과 MapReduce에 대한 흐름이 그림 2에 표시되어 있다.

Apache Hadoop 설정하기

이제 Linux VM에서 Apache Hadoop Cluster를 설정한 후 Apache Hadoop Cluster에서 MapReduce 애플리케이션을 실행한다.

Apache Hadoop은 다음과 같은 세 가지 개발 모드를 지원한다.

• 독립형 모드: 기본적으로 Hadoop은 비분산 독립형 모드로 실행하도록 구성한다. 이 모드는 애플리케이션을 디버그하기가 유용하다.
• 의사 분산 모드: 또한 Hadoop은 단일 노드 의사 분산 모드로 실행할 수 있다. 이 경우에 각 Hadoop 디먼은 개별 Java™ 프로세스로 실행된다.
• 완전 분산 모드: Hadoop은 다른 호스트에 구성되며 클러스터로 실행된다.

Hadoop을 독립형이나 의사 분산 모드로 설정하려면 Hadoop 웹 사이트의 참고자료를 참조한다. 이 기사에서는 완전 분산 모드에서 Hadoop을 설정하는 과정만을 다룬다.

환경 준비

이 기사에서는 세 대의 GNU/Linux 서버가 필요하며 한 대는 마스터 노드로 동작하며 다른 두 대는 슬레이브 노드이다.

각 시스템에 Hadoop 바이너리와 Java SE 6를 설치해야 한다. 자세한 정보는 참고자료 섹션을 참조한다. 이 기사에서는 Hadoop 버전 0.19.1을 사용한다.

또한 SSH를 설치하고 각 시스템에서 sshd를 실행해야 한다. SUSE 및 RedHat과 같은 일반적인 Linux 릴리스를 설치하면 이러한 것들이 기본적으로 설치된다.

통신 설정

/etc/hosts 파일을 갱신하고 IP와 호스트 이름을 사용하여 세 대의 시스템이 서로 연결되는지 확인한다.

Hadoop 마스터 노드는 SSH를 사용하여 슬레이브 노드와 통신하기 때문에 마스터 노드와 슬레이브 노드 간에 암호가 필요없는 인증된 SSH 연결을 설정해야 한다. 각 시스템에서 다음 명령을 실행하여 RSA 공용 키와 개인용 키를 생성한다.

	ssh-keygen –t rsa 

이렇게 하면 /root/.ssh 디렉토리 밑에 id_rsa.pub가 생성된다. 마스터 노드의 id_rsa.pub를 59_rsa.pub로 이름을 바꾸고 슬레이브 노드로 복사한다. 그런 다음 명령을 실행하여 마스터 노드의 공용 키를 슬레이브 노드의 인증된 키에 추가한다.

	cat /root/.ssh/59_rsa.pub >> /root/.ssh/authorized_keys 

이제 SSH를 사용하여 슬레이브 노드로 연결한다. 암호를 입력하지 않아도 연결이 되어야 한다.

마스터 노드 설정

<Hadoop_home>/conf/ 디렉토리 밑에 있는 해당 구성 파일을 구성하여 완전 분산 모드에서 작업하도록 Hadoop을 설정한다.

hadoop-site.xml 파일에서 Hadoop 전개를 구성한다. 이렇게 구성하면 hadoop-default.xml에 구성된 내용은 무시된다.

hadoop-site.xml

<?xml version="1.0"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<!-- Put site-specific property overrides in this file. -->
<configuration>
  <property>
    <name>fs.default.name</name>
    <value>hdfs://9.30.210.159:9000</value>
  </property>
  <property>
    <name>mapred.job.tracker</name>
    <value>9.30.210.159:9001</value>
  </property>
  <property>
    <name>dfs.replication</name>
    <value>1</value>
  </property>
  <property>
    <name>hadoop.tmp.dir</name>
    <value>/root/hadoop/tmp/</value>
  </property>
</configuration>

hadoop-env.sh 파일을 구성하여 JAVA_HOME을 지정한다. 해당 라인을 코멘트 아웃하고 JAVA_HOME 디렉토리를 지정한다.

	export JAVA_HOME=<JAVA_HOME_DIR>

마스터 노드의 IP 주소를 마스터 파일에 추가한다.

9.30.210.159

슬레이브 노드의 IP 주소를 슬레이브 파일에 추가한다.

	9.30.210.160
	9.30.210.161

슬레이브 노드 설정

SCP나 기타 복사 유틸리티를 사용하여 hadoop-site.xml, hadoop-env.sh, masters 및 slaves를 각 슬레이브 노드로 복사한다.

HDFS 포맷하기

다음 명령을 실행하여 Hadoop 분산 파일 시스템을 포맷하여 초기화한다.

	<Hadoop_home>/bin/hadoop namenode -format

Hadoop Cluster를 확인한다.

이제 bin/start-all.sh를 사용하여 Hadoop Cluster를 시작한다. 그러면 마스터 노드와 슬레이브 노드에 몇 가지 로그가 출력된다. 이 로그를 검증하여 모든 사항이 올바른지 확인한다. 오류가 있으면 HDFS를 포맷하여 hadoop-site.xml 파일에 지정된 Temp 디렉토리를 지우고 다시 시작한다.

다음 URL에서 마스터 노드와 슬레이브 노드가 이상이 없는지 확인한다.

NameNode: http://9.30.210.159:50070
JobTracker: http://9.30.210.159:50030

클라우드에서 Hadoop Cluster를 설정했으므로 이제 MapReduce 애플리케이션을 실행할 차례이다.

위로

MapReduce 애플리케이션을 작성한다.

MapReduce 애플리케이션은 "Map"과 "Reduce"의 특성을 갖고 있어야 하며 이는 작업을 더 작은 부분으로 나누어 병렬로 처리할 수 있다는 것을 의미한다. 그런 다음 각 하위 작업의 결과가 정리되어 원래의 작업에 대한 결과가 작성된다. 이에 대한 예로 웹 사이트 키워드 검색을 들 수 있다. 검색 및 그래빙 작업은 분할되어 슬레이브 노드에 위임되며 그 후에 각 결과가 집계되어 마스터 노드에서 최종 결과가 출력된다.

샘플 애플리케이션 실행하기

Hadoop에는 테스트용 샘플 애플리케이션이 있다. 이 중에는 여러 개의 파일에서 특정 단어의 빈도 수를 세는 단어 계수기가 있다. 이 애플리케이션을 실행하여 Hadoop Cluster를 확인한다.

먼저 conf/ 디렉토리 밑에 있는 input 파일을 분산 파일 시스템에 삽입한다. 그런 다음 이 파일에서 단어 수를 센다.

$ bin/hadoop fs –put conf input 

그런 다음 샘플 애플리케이션을 시작하여 "dfs"로 시작하는 단어의 빈도 수를 계산한다.

$ bin/hadoop jar hadoop-*-examples.jar grep input output 'dfs[a-z.]+'

이 명령은 Map과 Reduce 프로세스를 출력한다.

앞에 있는 두 명령을 실행하면 HDFS 밑에 "input"과 "output" 디렉토리가 생성된다. 다음 명령을 사용하면 이 디렉토리가 표시된다.

$ bin/hadoop fs –ls

분산 파일 시스템에서 출력된 파일을 보자. 이 파일에는 "dfs"로 시작하는 단어의 빈도 수가 키 값 쌍으로 표시되어 있다.

$ bin/hadoop fs -cat ouput/*

이제 JobTracker 사이트에서 완료된 작업 로그를 확인한다.

Log Analyzer MapReduce 애플리케이션 작성하기

이제 Hadoop의 WordCount 애플리케이션과 유사한 IBM WebSphere® Portal v6.0 Log Analyzer 애플리케이션을 작성한다. 이 분석기는 IBM WebSphere Portal v6.0의 모든 SystemOut*.log 파일을 분석하여 특정 기간에 이 Portal에서 애플리케이션이 실행된 횟수를 표시한다.

Portal 환경에서는 모든 로그가 5MB로 분할되어 몇 개의 노드에서 병렬로 분석된다.

public class PortalLogAnalyzer {
	
public static class Map extends MapReduceBase 
	implements Mapper<LongWritable, Text, Text, IntWritable> {

	private static String APP_START_TOKEN = "Application started:";
	 private Text application = new Text();
		
	 public void map(LongWritable key, Text value, 
		 OutputCollector<Text, IntWritable> output, 
		Reporter reporter) throws IOException {
			 
	    String line = value.toString();
	    if(line.indexOf(APP_START_TOKEN) > -1) {
		int startIndex = line.indexOf(APP_START_TOKEN);
		startIndex += APP_START_TOKEN.length();
		String appName = line.substring(startIndex).trim();
		application.set(appName);
		output.collect(application, new IntWritable(1));
	    }
	}
}
	
public static class Reduce extends MapReduceBase 
	    implements Reducer<Text, IntWritable, Text, IntWritable> {
		
	public void reduce(Text key, Iterator<IntWritable> values, 
		OutputCollector<Text, IntWritable> output, 
		Reporter reporter) throws IOException {
	
	    int sum = 0;
	    while(values.hasNext()) {
		sum += values.next().get();
	    }
	    output.collect(key, new IntWritable(sum));
	}
}
	
public static void main(String[] args) throws IOException {
	JobConf jobConf = new JobConf(PortalLogAnalyzer.class);
	jobConf.setJobName("Portal Log Analizer");
	jobConf.setOutputKeyClass(Text.class);
	jobConf.setOutputValueClass(IntWritable.class);
	jobConf.setMapperClass(Map.class);
	jobConf.setCombinerClass(Reduce.class);
	jobConf.setReducerClass(Reduce.class);
	jobConf.setInputFormat(TextInputFormat.class);
	jobConf.setOutputFormat(TextOutputFormat.class);
		
	FileInputFormat.setInputPaths(jobConf, new Path(args[0]));
	FileOutputFormat.setOutputPath(jobConf, new Path(args[1]));
	JobClient.runJob(jobConf);
}
}

Hadoop API에 대한 자세한 설명은 Hadoop 사이트의 API 문서를 참조한다. 다음은 간단한 설명이다.

Map 클래스에서는 로그 파일의 각 라인을 분석하여 애플리케이션의 이름을 가져오는 맵 함수를 구현한다. 그런 다음 애플리케이션의 이름을 키 값 쌍으로 출력 콜렉션에 삽입한다.

Reduce 클래스는 키 또는 애플리케이션 이름이 동일한 모든 값을 합한다. 그러면 이 애플리케이션은 Portal에서 각 애플리케이션이 시작된 횟수를 표시하는 키 값 쌍을 출력한다.

Main 함수는 MapReduce 작업을 구성하고 실행한다.

PortalLogAnalyzer 실행하기

먼저, Java 코드를 마스터 노드로 복사한 후 이 코드를 컴파일한다. Java 코드를 <hadoop_home>/workspace 디렉토리로 복사한다. 이 코드를 컴파일하여 나중에 Hadoop 명령을 사용하여 실행할 수 있도록 Jar 파일로 아카이브한다.

$ mkdir classes
$ javac –cp ../hadoop-0.19.1-core.jar –d classes
			hadoop/sample/PortalLogAnalyzer.java
$ jar –cvf PortalLogAnalyzer.jar –C classes/ .

Portal의 로그를 workspace/input으로 복사한다. 2009년 5월의 모든 로그를 포함하고 있는 몇 개의 로그 파일이 있다고 가정한다. 이 로그를 HDFS에 가져다 놓는다.

$ bin/hadoop fs –put workspace/input input2

PortalLogAnalyzer를 실행하면 출력에 Map과 Reduce의 프로세스가 표시된다.

$ bin/hadoop jar workspace/PortalLogAnalizer.jar hadoop.sample.PortalLogAnalizer input2 
     output2

애플리케이션이 종료하면 아래 그림 4와 같은 내용이 출력된다.

$ bin/hadoop fs –cat output2/*

JobTracker 사이트에서 또 다른 완료된 작업을 확인할 수 있다. 그림 5의 마지막 라인에 유의한다.