정신 없이 한달을 보냈다.
분위기 적응에 이것저것 잡다한 일들이 많아 몸도 마음도 모두 힘들었는데,
지금은 마음의 여유가 조금 생겼다.

그만큼 뻔뻔해진 것이겠지만 ^^;

이제 조금만 더 열심히 해보자.
항상 문제는 내 안에 있는 것이니까.

그리고 아래는 내 머리속 깊이 박혀있던 불만을 날려버린
아주 중요한 문장이다. 오늘 우연히 보게 되었는데 정말정말 다행이다.

In science the credit goes to the man who convinces the world,
not to the man to whom the idea first occurs.

- Sir William Osler-

이십년이 흐른 후

당신은 이룬 일들보다

하지 못한 일들로 인해 더 깊이 좌절하리라.

설치는 그다지 어렵지 않았지만 의외로 테스트 프로그램 작성 및 실행에 시간이 걸렸다.
기초를 다시 다지는 기회가 되었음.

설치환경 : VMWare Server + JDK 1.6 + Hadoop 0.18.1
설치버전 : HBase 0.18.0

1. Download

http://www.apache.org/dyn/closer.cgi/hadoop/hbase/
http://apache.tt.co.kr/hadoop/hbase/hbase-0.18.0/hbase-0.18.0.tar.gz

2. Installation

[hadoop@master hadoop]$ pwd
/home/hadoop
[hadoop@master hadoop]$ tar xzvf hbase-0.18.0
[hadoop@master hadoop]$ ln -s ./hbase-0.18.0 hbase

3. Configuration

[hadoop@master conf]$ pwd
/home/hadoop/hbase/conf
[hadoop@master conf]$ vi hbase-env.sh
==> JAVA_HOME 수정
[hadoop@master conf]$ vi hbase-site.xml
<configuration>
    <property>
        <name>hbase.rootdir</name>
        <value>hdfs://localhost:9000/hbase</value>
    </property>
</configuration>

4. Test HBase Shell

[hadoop@master hbase]$ pwd
/home/hadoop/hbase
[hadoop@master hbase]$ export HADOOP_HOME=/home/hadoop/hadoop
[hadoop@master hbase]$ $HADOOP_HOME/bin/start-all.sh
.....
[hadoop@master hbase]$ bin/start-hbase.sh
starting master, ...
localhost: starting regionserver, ...
[hadoop@master hbase]$ bin/hbase shell
HBase Shell; enter 'help<RETURN>' for list of supported commands.
Version: 0.18.0, r697626, Sun Sep 21 16:00:50 PDT 2008
hbase(main):001:0> create 'myTable', { NAME=>'myColumnFamily' }
0 row(s) in 6.6440 seconds
hbase(main):001:0> list
myTable

1 row(s) in 0.0490 seconds
hbase(main):001:0> quit

5. Test an Example Program
Original program source : http://hadoop.apache.org/hbase/docs/current/api/overview-summary.html#overview_description

sample/MyClient.java

package sample
import java.io.IOException;

import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Scanner;
import org.apache.hadoop.hbase.io.BatchUpdate;
import org.apache.hadoop.hbase.io.Cell;
import org.apache.hadoop.hbase.io.RowResult;

public class MyClient {

 public static void main(String[] args) throws IOException {
  // 클라이언트에서 어디로 접속할지 설정 객체에게 알려야 한다.
  // 기본으로 로컬 설정 파일에서 해당 정보를 가지고 오니 걱정하지는 말자.
  HBaseConfiguration config = new HBaseConfiguration();
 
  // 'myTable' 테이블에 접속하는 HTable 객체를 생성한다.
  HTable table = new HTable(config, "myTable");
 
  // row에 대한 모종의 업데이트를 수행하기 위해서는 BatchUpdate 객체를 사용한다.
  // BatchUpdate는 row 하나와 옵션으로 timestamp 값을 받는다.
  BatchUpdate batchUpdate = new BatchUpdate("myRow");
 
  // BatchUpdate의 put() 메소드는 값을 넣은 셀을 기술하는 텍스트와
  // 저장하고자 하는 값이 있는 byte 배열을 인자로 받는다.
  // 따라서 String이나 다른 primitives등을 저장하고자 한다면 getBytes()를 사용해야한다.
  batchUpdate.put("myColumnFamily:columnQualifier1", "columnQualifier1 value!".getBytes());
 
  // HBase에서는 삭제도 배치 작업이다.
  //batchUpdate.delete("myColumnFamily:cellWantDeleted");
 
  // 작업을 끝낸 뒤에는 결과를 commit 한다.
  // HTable#commit 메소드는 BatchUpdate 객체를 인자로 받아 배치작업을 수행한다.
  table.commit(batchUpdate);
 
  // 이제 방금 저장한 값들을 읽어보자. 리턴되는 값은 Cell 객체이다.
  // Cell은 byte 배열로 이루어진 값과 timestamp 값의 조합이다.
  Cell cell = table.get("myRow", "myColumnFamily:columnQualifier1");
  String valueStr = new String(cell.getValue());
 
  // 찾고자 하는 row를 모를 경우, Scanner를 사용한다. Scanner는 테이블의 내용에 대한
  // cursor 스타일의 인터페이스를 제공한다.
  Scanner scanner = table.getScanner(new String[]{"myColumnFamily:columnQualifier1"});
 
  // HBase 0.2에 있는 Scanner는 RowResult 객체를 리턴한다.
  // RowResult는 row key와 column들을 하나의 인터페이스로 묶어놓은 것과 같다.
  // RowResult#getRow는 row key를 리턴한다.
  // 또한 RowResult는 Map 인터페이스를 구현해두고 있기 때문에 column 결과에 쉽게 접근 가능하다.
  RowResult rowResult = scanner.next();
  while (rowResult != null) {
   System.out.println("Found row: " + new String(rowResult.getRow()) + " with value: " + rowResult.get("myColumnFamily:columnQualifier1".getBytes()));
   
   rowResult = scanner.next();
  }
 
  // iterable한 Scanner를 foreach로 탐색하는 방법도 있다.
  for (RowResult result : scanner) {
   System.out.println("Found row: " + new String(result.getRow()) + " with value: " + result.get("myColumnFamily:columnQualifier1".getBytes()));
  }
 
  // 작업이 끝나면 Scanner를 닫도록 한다.
  scanner.close();
 }
}

build.xml

<project name="MyClient">
 <property name="target.dir" value="build" />
 <property name="classes.dir" value="${target.dir}/classes" />
 
 <property name="source.dir" value="src" />
 <property name="lib.dir" value="lib" />
 
 <target name="init">
  <mkdir dir="${classes.dir}" />
  <path id="build.classpath">
   <fileset dir="${lib.dir}">
    <include name="**/**.jar" />
   </fileset>
  </path>
 </target>
 
 <target name="compile" depends="init">
  <javac
   srcdir="${source.dir}"
   destdir="${classes.dir}"
   classpathref="build.classpath"
   encoding="utf-8"
  />
 </target>
</project>

[hadoop@master example]$ ls
build.xml   lib   src
[hadoop@master example]$ ant compile
...
[hadoop@master example]$ ls
build    build.xml    lib    src
[hadoop@master example]$ cd build/classes
[hadoop@master example]$ ls
sample
[hadoop@master classes]$ export HBASE_HOME=/home/hadoop/hbase
[hadoop@master classes]$ $HBASE_HOME/bin/hbase sample.MyClient
Found row: myRow with value: timestamp=1217998267705, value=columnQualifier1 value!


이렇게 정리해두면 언젠가 도움이 되겠지...ㅠ

Hadoop Streaming을 이용해 Python으로 만들어진 mapper와 reducer를 구동해보았다.

환경 : VMWare Server + CentOS 5.1 + Hadoop 0.18.1 + JDK 1.6

mapper.py

#!/usr/bin/python

import sys

def mapper():
        # standard input
        for line in sys.stdin.readlines():
                words=line.split(' ')
                for word in words:
                        print '%s\t%d'%(word,1)

if __name__=='__main__':
        mapper()

reducer.py

#!/usr/bin/python

import re
import sys
import logging

filter=re.compile(r'[,.!<>@#$%&*"]',re.U)

def reducer():
        last_key=''
        count=0

        for line in sys.stdin.readlines():
                try:
                        line=filter.sub('',line.lower())

                        words=line.split('\t')
                        key=words[0]
                        val=words[1]

                        if key!=last_key:
                                print '%s\t%d\n'%(last_key,count)
                                count=int(val)
                        else:
                                count = count + int(val)

                        last_key=key
                except:
                        logging.error('an error occured while parsing the line.')

if __name__=='__main__':
        reducer()

.bash_profile 수정

HADOOP_HOME=/home/hadoop/hadoop
HSTREAMING="$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/contrib/streaming/hadoop-0.18.1-streaming.jar"
export HADOOP_HOME HSTREAMING

실행

 $HSTREAMING -mapper "./mapper.py" -file mapper.py -reducer "./reducer.py" -file reducer.py -jobconf mapred.reduce.tasks=10 -jobconf mapred.map.tasks=10 -input "page" -output "./log_result"

파이썬을 사용하면서 가장 불만인 것이 바로 헷갈리는 한글 인코딩이었다.
틈만나면 깨지는 한글을 보면서 한숨만 내쉴수는 없는 일...
경험으로 알게된 몇가지를 정리해보자.
(틀릴수도 있으니 혹시나 아시는 분께서는 정정부탁드립니다 ㅠ__ㅠ 지식이 미천해서..)

1. 초기상태의 파이썬은 기본 인코딩은 ASCII 이다.

import sys
sys.getdefaultencoding()

으로 알 수 있다.

2. 소스파일 인코딩
파이썬 소스파일의 앞에 아래의 주석을 다는 것은,

# -*- coding: utf-8 -*-

이 소스파일이 utf-8 인코딩으로 저장된 파일이라는 뜻이다.

3. str 타입 vs. unicode 타입
str 타입은 파이썬 인터프리터가 실행된 환경의 로케일에 따라 변환된 문자열이고,
unicode 타입은 문자열이 로케일에 상관없이 통일된 문자체계를 가지는 코드로 변환된 것이다.

>>> type('파이썬')
(type 'str')
>>> type(u'파이썬')
(type 'unicode')

4. 인코딩 vs. 디코딩
파이썬에서는 유니코드가 기본 코드인 것 같다. 따라서
유니코드 --> 다른 CharacterSet : 인코딩(encode)
다른 CharacterSet --> 유니코드 : 디코딩(decode)
이러한 공식이 성립한다.
이를 확인하기 위해 다음과 같이 해볼 수 있다. (윈도우 환경-cp949-이다)

>>> s='파이썬'
>>> u=u'파이썬'
>>> s==u
False
>>> s.decode('cp949')==u
True
>>> s==u.encode('cp949')
True

갑자기 멀티쓰레드 서버가 필요하여 구현을 해보려다
좋은 article을 보게 되어 정리해둔다.

출처 : http://www.devshed.com/c/a/Python/Sockets-in-Python-Into-the-World-of-Python-Network-Programming/3/

[개요]
1. 각 연결에 대한 쓰레드를 생성
멀티쓰레드 서버의 핵심. 수락된(accepted) 각 연결 요청에 대해 새로운 쓰레드가 생성되고
서로 독립된 쓰레드가 개별 클라이언트에게 서비스를 제공한다.

2. 핸들러 생성
핸들러에서는 실제 작업이 수행된다.

[코드]
from socket import *
from threading import *

def handler(clientsock,addr):
       while 1:
             data = clientsock.recv(BUFSIZ)
             if not data: break
                   clientsock.send(‘echoed:..’, data)
 
       clientsock.close()

if __name__==’__main__’:
       HOST = 'localhost'
       PORT = 21567
       BUFSIZ = 1024
       ADDR = (HOST,PORT)
       serversock = socket(AF_INET,SOCK_STREAM)
       serversock.bind(ADDR)
       serversock.listen(2)

       while 1:
             print 'waiting for connection...'
             clientsock, addr = serversock.accept()
             print '...connected from:', addr
             thread.start_new_thread(handler,(clientsock, addr))

       #some other cleanup code if necessary

http://www.freesearch.pe.kr/941

오랜만의 포스팅이 따라하기(?)라니 좀 슬프지만 :) 그래도 한번 해보았다.
한RSS에서 제공하는 RSS의 구독자수를 크롤링해서 순위를 메겨보는 것이 목표.

고감자님과 같이 Python과 BeautifulSoup, feedparser를활용하여 비교적 쉽게 구현할 수
있었지만, 사실 생각보다 쫌 오래걸렸다ㅠㅠ
(고감자님은 40라인으로 해결하셨다는데 난 80라인 정도라 부끄러워 코드는 못올리겠다;;)

나는 한RSS에서 볼 수 있는 테마별 분류 리스트를 이용해서 얻을 수 있는 모든 RSS 피드를
가지고 왔는데, 테마 분류는 별로 필요가 없을 듯 하여 크롤링 결과에 포함시키지 않았다.
이렇게 해서 총 1783개의 RSS 피드와 구독자수를 얻을 수 있었다.

고감자님의 자료만큼 신뢰도가 높지는 않겠지만,
이런 식으로 데이터를 얻어낼 수도 있다는 사실을 알게 된 좋은 경험이었다.
(훌륭한 과제를 내주신 고감자님께 감사드립니다^^)

마지막으로 크롤링한 데이터와 이를 분석하여 엑셀파일로 정리한 자료를 첨부한다.

Beautiful Code의 23장 MapReduce를 이용한 분산 프로그래밍 내용 중에,
word count를 처리하기 위한 몇가지 시도들이 소개된다.
사용할 기회가 없는 C++을 까먹지 않기 위해 그 중 예제 23-1, 2, 3을 구현해보았다.
(4번은 너무 어려워서 ㅠ__ㅠ 누가 가르쳐 주실 분은 안계시려나..)

그나저나 어째서 파일이름을 MapReduce로 했을까;; 고치기 귀찮으니 일단 패스...

구현 환경은 Eclipse CDT 3.3 + Cygwin + GCC 3.4

예제 23-1. 소박한 직렬 단어 집계 프로그램

예제 23-2. 병렬 단어 집계 프로그램

예제 23-3. 분할된 저장소를 사용하는 병렬 단어 집계 프로그램

한동안 손대지 못했던 하둡을 다시 설치해보다 서버 설정에서 약간의 삽질을 했다.
하둡의 문제라기 보다는 편한 운용을 위한 ssh 설정의 문제였음.

1. Hadoop을 실행할 계정과 그룹은 hadoop/hadoop으로 한다.

$ sudo addgroup hadoop
$ sudo adduser --ingroup hadoop hadoop

2. Hadoop 스크립트가 ssh를 통해 해당 계정으로 자유롭게 접속할 수 있도록 설정한다.

hadoop@cluster01:~$ ssh-keygen -t rsa -P ""
Generating public/private rsa key pair.
Enter file in which to save the key (/home/hadoop/.ssh/id_rsa): <엔터입력>
...
hadoop@cluster01:~$ cd .ssh
hadoop@cluster01:~$ cat .ssh/id_rsa.pub >> .ssh/authorized_keys
hadoop@cluster01:~$ ssh localhost
The authenticity of host 'localhost (127.0.0.1)' can't be established.
RSA key fingerprint is ....
Are you sure you want to continue connecting (yes/no)? <yes 입력>
Warning: Permanently added 'localhost' (RSA) to the list of known hosts.
...

3. 이 때 암호가 설정되어 있지 않은 공개키를 기반으로 자동 접속이 되어야 하는데
자꾸 암호를 물어오는 경우가 있다. 약간의 삽질 끝에 해결한 방법은 다음과 같다.

일단 이러한 문제의 원인으로 발견한 것은 아래의 두 가지이다.
(1) /etc/ssh/sshd_config 설정 문제
(2) .ssh/authorized_keys의 permission 문제

(1)번 문제는 다음과 같이 조치한다.

/etc/ssh/sshd_config 내용에서 다음과 같은 부분을 찾아 고친 후 sshd를 재시작한다.
...
PubkeyAuthentication yes
AuthorizedKeysFile .ssh/authorized_keys
...
PermitEmptyPasswords yes

hadoop@cluster01:~$ /etc/init.d/sshd restart

(2)번 문제가 발생하면 접속 시 다음과 같은 메시지를 볼 수 있다.

Permission denied (publickey,gssapi-keyex,gssapi-with-mic,keyboard-interactive)

이는 .ssh/authorized_keys의 group permission에 write 권한이 있는 경우로,
다음과 같이 조치한다.

hadoop@cluster01:~$ chmod g-w .ssh/authorized_keys

위 두 문제를 해결하면 ssh에 관련된 오류가 사라질 것이다.