대한민국에서 Programmer로 살아가기

1. vmware 에서 가상머신에 centos를 설치한다.
   hostname을 zk001로 한다.
   
   ** 호스트 네임 변경방법
   "/etc/sysconfig/network" 파일을 수정후 재부팅한다.
   
   NETWORKING=yes
   HOSTNAME=zk001
   
2. root로 openjdk 1.8.0을 설치한다.
   $> yum -y install java-1.8.0-openjdk
3. iptables 서비스를 종료한다.
   $> service iptables stop
4. 서비스 자동시작을 해제한다.
   $> chkconfig iptables off   
   
5. 위의 가상머신을 clone 기능을 이용해 복제 하고
   hostname을 kafka001로 한다. 

   ** 호스트 네임 변경방법
   "/etc/sysconfig/network" 파일을 수정후 재부팅한다.
   
   NETWORKING=yes
   HOSTNAME=kafka001
   
6. zk001서버에 주키퍼를 설치한다.
    $>cd /usr/local/
    $>wget https://archive.apache.org/dist/zookeeper/zookeeper-3.5.5/apache-zookeeper-3.5.5-bin.tar.gz
    - 압축을 해제한다.
$>tar xvfz apache-zookeeper-3.5.5-bin.tar.gz
- 소프트 링크를 생성한다.
$>ln -s apache-zookeeper-3.5.5-bin zookeeper
$>mkdir /data

7. zk001를 clone 기능을 이용하여 2대 복제하고 hostname을 zk002, zk003으로 수정한다.
   zk001 서버에서 /data/myid에 1을 입력한다.
   $> echo 1 > /data/myid
   zk002 서버에서 /data/myid에 2을 입력한다.
   $> echo 2 > /data/myid
   zk003 서버에서 /data/myid에 3을 입력한다. 
   $> echo 3 > /data/myid
   
8. zk001, zk002, zk003 서버에서 환경설정파일을 생성한다.
$> cd /usr/local/zookeeper/conf
$> vi zoo.cfg

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/data
clientPort=2181
server.1=zk001:2888:3888
server.2=zk002:2888:3888
server.3=zk003:2888:388

9. zk001, zk002, zk003  서버에서 주키퍼가 정상적으로 수행되는지 확인한다.
   $> cd /usr/local/zookeeper/bin/
   $> ./zkServer.sh start
   
   아래 명령어로 daemon이 떠있는지 확인한다.
   $>ps -ef | grep zookeeper
   
10. 5번 step에서 복제한 kafka001에 kafka를 설치한다.
    $> cd /usr/local/  
    $> wget https://archive.apache.org/dist/kafka/2.1.0/kafka_2.11-2.1.0.tgz
    $> tar xvfz kafka_2.11-2.1.0.tgz
    $> ln -s kafka_2.11-2.1.0 kafka
$> mkdir /data1
$> mkdir /data2

11. kafka001 서버를 2대 복제하고 hostname을 kafka002, kafka003으로 변경한다.
    각서버에서 주키퍼 서버로 접속이 되는지 확인한다.
$> nc -v zk001 2181
Connection to zk001 2181 port [tcp/eforward] succeeded!

12. 각서버에서 환경설정 파일을 변경한다.
    $> cd /usr/local/kafka/config/
vi 에디터로 아래 파일을 수정한다.
$> vi server.properties
  아래 부분을 찾아서 서버에 맞게 수정해 준다.
  ############################# Server Basics #############################
      # The id of the broker. This must be set to a unique integer for each broker.
      broker.id=1

  kafka001은 1로 수정
  kafka002는 2로 수정 
  kafka003은 3으로 수정
  
13. log 디렉토리 설정 수정
    아래 부분을 찾아서 각 서버에서 동일하게 수정을 한다.
log.dirs=/data1,/data2  
    
14. zookeeper.connect부분을 찾아서 아래와 같이 모든 서버에서 수정해줍니다.
    아래와 같이 입력하면 kafka가 주키퍼에 접속하여 /kafka라는 znode를 생성합니다.

    zookeeper.connect=zk001:2181,zk002:2181,zk003:2181/kafka

15. 각 서버에서 카프카를 시작한다 이때 & 기호를 붙여서 백그라운드에서 실행이 되도록 합니다.

   ./kafka-server-start.sh /usr/local/kafka/config/server.properties &

16. 주키퍼와 카프카의 상태확인
    zk001,zk002,zk003 서버에서 주키퍼 클라이언트 포트의 상태를 확인합니다.

$> ps -ef | grep 2181
root     12758  2653  0 23:09 pts/0    00:00:00 grep 2181

kafka001,kafka002,kafka003 서버에서 kafka 포트의 상태를 확인합니다.

$> ps -ef | grep 9092
root      9126  2459  0 23:09 pts/0    00:00:00 grep 9092

17. 주키퍼와 cli를 통해 주키퍼와 카프카가 잘연결 되었는지 확인
    $> cd /usr/local/zookeeper/bin
    $> ./zkCli.sh
[zk: localhost:2181(CONNECTED) 0] ls /
    [kafka, zookeeper]

[zk: localhost:2181(CONNECTED) 1] ls /kafka/brokers/ids
    [1,2,3]


18. 카프카 토픽 생성
$>  cd /usr/local/kafka/bin/
$> ./kafka-topics.sh \
--zookeeper zk001:2181,zk002:2181,zk003:2181/kafka \
--replication-factor 1 --partitions 1 --topic kang-topic --create

   - 참고로 토픽을 잘못 생성한 경우는 아래와 같이 삭제를 한다.
$> ./kafka-topics.sh \
--zookeeper zk001:2181,zk002:2181,zk003:2181/kafka \
--topic kang-topic --create

19. 카프카 메세지 퍼블리싱
    아래 명령어를 수행하면 메세지를 입력할 수 있는 프롬프트가 뜹니다.
    $> cd /usr/local/kafka/bin/ 
$> ./kafka-console-producer.sh \
   --broker-list kafka001:9092,kafka002:9092,kafka003:9092 \
   --topic kang-topic
   
프롬프트에 아래와 같이 입력합니다.
>first message
>second message

메세지 입력후 ctrl+c를 입력하면 입력창에서 나올 수 있습니다.

20. 카프카 메세지 수신
    아래와 같이 입력하면 kang-topic에 퍼블리싱된 메세지를 아래와 같이 확인할 수 있다.
$>./kafka-console-consumer.sh \
  --bootstrap-server kafka001:9092,kafka002:9092,kafka003:9092 \
  --topic kang-topic --from-beginning

    first message
    second message

hive 타입 사용 시 주의 사 

정수 및 소수(int,bigint, double,float)
   - Insert 시 값에 싱글 쿼테이션으로 감싸도 입력가능하나  null로 입력 되는 경우도 있음(버전마다 다름)
   - Select시 where절에서 비교시 싱글 쿼테이션으로 감싸도 조회가 가능

date, timestamp
   - insert, select시에 모드 싱글쿼테이션으로 감싸 줘야 한다.

decimal  
   - default는 10,0 자리임 소수부를 입력하고 싶으면 필히 소수부를 지정해야 함.
   - 입력값이 소수부 자릿수 보다 큰 경우 정의된 소수자릿에서 4사5입됨
   - 입력값이 정수부 자릿수 보다 큰 경우 null로 입력됨

char, vachar 
   - 지정된 자릿수를 넘는 값을 입력 시에 정의된 자릿수 까지만 잘려서 입력됩니다.

/*select count(*) from employees*/
>db.employees.aggregate([{$group:{_id:null, count:{$sum:1}}}])

/*select sum(sal) from employees*/
>db.employees.aggregate([{$group:{_id:"$deptno",total:{$sum:"$sal"}}}])

/*select deptno,sum(sal) from employees where sal=3000 group by deptno*/
>db.employees.aggregate([{$match:{sal:3000}}, {$group:{_id:"deptno", total:{$sum:"$sal"}}}])

/*deptno가 10인 데이터를 조회*/
>db.employees.find({deptno:{$all:[10]}},{_id:0, empno:1,ename:1,sal:1})

/*deptno가 10이고 deptno가 30인 데이터를 조회*/
>db.employees.find({deptno:{$all:[10,30]}},{_id:0, empno:1,ename:1,sal:1})

>db.employees.find({$and:[{deptno:10},{deptno:30}]},{_id:0, empno:1,ename:1,sal:1})

/*deptno가 10이거나 deptno가 30인 데이터를 조회*/
>db.employees.find({deptno:{$in:[10,30]}},{_id:0, empno:1,ename:1,sal:1})

>db.employees.find({$or:[{deptno:30},{deptno:10}]},{_id:0, empno:1,ename:1,sal:1})

/*comm field가 존재하는 데이터만 조회하라*/
db.employees.find({comm:{$exists:true}},{_id:0,empno:1,ename:1})

1.emp collection 생성
>for (var i=1; i<=20; i++) db.emp.save({empno:i, empname:"test", sal:100})
>db.emp.find()

2.배열에 결과저장 배열에 저장되어 있는 내용출력
>var cur = db.emp.find()
>while(cur.hasNext()) printjson(cur.next())

3.배열의 특정 순번에 있는 내용출력
>var cur = db.emp.find()
>printjson(cur[10])

4.Date함수를 통해 현재 날짜 출력
>x=new Date()

5.x변수에 저장된 날짜를 문자열로 출력
>x.toString()

6.ISODate 함수를 통해 현재 날짜를 출력
>d=ISODate()
>d.getMonth()
>d.getYear()
>d.getDay()

7.Sequence Number 함수
>Function seq_no("name"){
  var ret=db.seq_no.findAndModify(query:{_id:name},
                                             update:{$inc:{next:1}},
                                             "new"true, upsert:true});
  return ret.next;
  }
원리는 seq_no라는 함수형 collection을 만들어서 해당 _id 별로
+1씩 증가시킨 값을 리턴한다.

8.selec 문
/*select * from employees where empno=7369 */
>db.employees.find({empno:7369}).forEach(printjson)

/*select ename from employees where empno=7900*/
첫번째 인자 {empno:7900}은 where절이다.
두번째 인자 {ename:""}은 select 절이다.
>db.employees.find({empno:7900}, {ename:""}).forEach(printjson)

/*select ename,job from employees where empno=7900*/
>db.employees.find({empno:7900}, {ename:"", job:""}).forEach(printjson)

9.index 생성문과 select  문
>db.employees.ensureIndex({ename:1})
/*select empno, ename from employees where ename>='ALLEN' and ename<= 'SCOTT';*/
>db.employees.find({}, {_id:0, empno:1, ename:1}).min({ename:"ALLEN"}).max({ename:"SCOTT"})

/*SELECT EMPNO, ENAME, HIREDATE FROM EMPLOYEES WHERE DEPTNO>=20 AND DEPTNO<=30*/
>db.employees.ensureIndex({deptno:1})
>db.employees.find({$min : {deptno:20}, $max: {deptno:30}, $query:{}}, {_id:0,empno:1,ename:1,hiredate:1})

/*select empno, ename from employees where empno>7500 and empno<=7600*/
>db.employees.find({empno:{$gt:7500, $lte:7600}},{_id=0,empno:1,ename:1})

/*select empno, ename, from employees where empno=7782 or empno=7844*/
>db.employees.find({$or:[{empno:7782}, {empno:7844}]},{_id=0, empno:1, ename:1})

/*select count(*) from employees:*/
>db.employees.count()

/*select count(*) from employees where empno>7900*/
>db.employees.find({empno:{$gt:7900}}).count()

/*select distinct deptno from employees;*/
>db.employees.distinct("deptno");

/*select deptno, sum(sal) csum
    from employees
  where hiredate>="01-01-1980" and hiredate <= "31-12-1981"
  group by deptno*/
>db.employees.group({key: {deptno:true},
                            cond: {"hiredate": {$gte: "01-01-1980", $lt:"31-12-1981"}},
                            reduce: function(obj,prev) {prev.csum += obj.sal;},
                            initial:{csum:0}
  })

/*select ename, job from employees where deptno=10 order by ename desc*/
db.employees.find({deptno:10},{_id:0,ename:1,job:1}).sort(ename:-1)

1. 데이터베이스가 생성될 물리적 경로생성
c:\>mkdir c:\MONGODB\test

2. mongoDB 버전 확인
>mongod --version

3. mongoDB 인스턴스 활성화
>mongod --dbpath c:\mongodb\test
-- mogodb shell프로그램 실행
>mongo

4.test 데이터베이스로 이동할때
>use test

5. 데이터베이스 목록 출력
>show dbs

6. 데이터베이스 상태확인
>db.stats()

7. 데이터베이스 shutdown
admin 영억으로 이동후에 셧다운 해야함.
>use admin 
>db.shutdownServer()

8. 데이터베이스 로그아웃
>db.logout()

9.collection 생성
capped:true이면 해당 익스텐트를 모두 사용하게되면
처음부터 재 사용할 수 있는 데이터 구조를 생성할 때
size 해당 Collection의 최초 생성크기
>db.createCollection("emp",{capped:false, size:8192});

10. colection list보기
show collections

11. collection의 현재상태 및 정보분석
>db.emp.validate();

12.collection의 이름변경
>db.emp.renameCollection("employee")

13.Collection의 삭제
>db.employee.drop();

14.collection에 데이터 INSERT
>db.emp.insert({eno:1101,fname:"JIMMY"});

15.collection에 데이터 UPDATE
>db.emp.update({eno:1101},{$set:{fname:"JOO"}});

16.collection에 데이터 SELECT
>db.emp.find().sort({eno:-1});

17.도큐먼트(row)부터 정의하고 collection 생성
>m={ename :  "smith"}
>n={ename :  1101}
>db.things.save(m)
>db.things.save(n)
>db.things.find()
>db.things.insert({ empno : 1102, ename:"king"})

18.for문을 이용한 증감된 값을 Collection에 insert
>for(var n=1103; n<=1120; n++) db.things.save({n:n, m:"test"+n})

19.db.things.find()로 조회시 리스트가 20 row가 넘는 경우 다음 페이지를 보고싶을때
>it

NOSQL이란?

No SQL
Not ONLY SQL
Non-Relational Operation Database SQL
RDBMS는 클라우드 컴퓨팅 환경에서 발생하는 빅데이터를 효과적으로
저장, 관리하는데 여러 가지 문제점이 발생하고 있는 상황에 직면
이런 문제점을 보완하기 위해 새로운 저장기술이 필요하게됨
이러한 배경에서 태어난것이 NoSQL

nosql의 장점 및 단점

하드웨어 확장에 제약이 없다.
저렴한 비용으로 분산 병렬처리가 가능하다.
메모리 mapping기능을 통해 읽고 쓰기가 빠르다.
비정행 데이터구조로 설계비용이 감소한다.
join인 없는 구조로 simple하고 성능이 빠르다.
기존의 하드웨어를 활용하여 구축할 수 있다.
RDBMS를 사용한 많은 시간이 흐르면서 성능/관리면에서 안정기에 접어들었다면
NOSQL 아직 성장하고 있는 기술이기에 문제점도 존재하는 것이 사실이다.
하지만 빅데이터를 처리할 수 있는 새로운 요구와 맞물려 빠른 속도로 성장 할 것으로 예측됨
RDBMS가 클라이언트/서버환경에 맞는 데이터 저장기술이라면
NOSQL은 클라우드 환경에 맞는 데이터 저장기술이다.
NoSQL이 RDBMS를 완전 대체하기보다는 상호 보완적인으로
발전할 것으로 예상됨.

NoSQL의 종류

대략 150 가지 이상의 제품이 존재함.
하지만 실제 사용자층을 확보하고 있고 꾸준한 기술지원과 관리를 지원하는제품은
대략 6가지 정도가 있다.
MongoDB, Casandra, HBASE, CouchDB, Redis

NoSQL 제품 유형
1.key-value Database (Riak, Voldemort)
2.Document Database (mongoDB, CouchDB)
3.Big Table Database (Hbase, Casandra)
4.Graph Database (Sones,AllegroGraph)

flume 이란 ?

FLUME ARCHITECTURE

master : 저장소 + Flume master
Agent : 데이터를 생산하는 시스템에 설치.
Collector : 에이전트로 부터 데이터를 수집하여 HDFS에 보내는 역할
( 에이전트와 콜렉터 상태 모니터링 )
에이전트가 발생한 로그나 데이터들을 모니터링하여 데이터 스트림을 생성하여 콜렉터로 보낸다.
콜렉터는 작은 스트림들을 묶어서 보다 큰 스트림으로 뭉친 다음 HDFS에 저장한다

하둡을 설치한 머신에 flume을 같이 설치하는 경우에는 hadoop namenode 포트를 변경해 줘야 한다.
왜 그런지 모르겠으나 flume 설치후 namenode가 시작이 안된다.

그리고 flume설치후 리부트를 하면 flume master 데몬은 띄우면 안된다.
자동 실행되기때문에 재 실행하면 떠있는 데몬이 죽는다.

flume 다운로드
 http://archive.cloudera.com/cdh4-deprecated/cdh/4/flume-0.9.4-cdh4.0.1.tar.gz
압축해제
tar xvfz flume-0.9.4-cdh4.0.1.tar.gz
환경변수 설정
gedit /etc/profile
export FLUME_HOME=/usr/lib/flume
export FLUME_CONF_DIR=/usr/lib/flume/conf

flume-site.xml 수정
cd /home/hadoop/flume/conf
vi flume-site.xml
<configuration>
  <property>
    <name>flume.master.servers</name>
    <value>flumemaster</value>
    <description>This is the address for the config servers status
 server (http)
    </description>
  </property>
</configuration>

flume master 실행
$ ./bin/flume master
flume을 최초에 실행하고 나서 리부트 하면
자동으로 Flume 데몬이 뜨므로 다시 띄우면 안됨
다시 띄우면 데몬이 죽고 다시 실행이 안됨
http://localhost:35871/

Flume Agent 실행
새로운 터미널을 열어서 아래 행을 실행한다.
$ ./bin/flume node_nowatch -n agent01
http://localhost:35862/

Flume Collector
새로운 터미널을 열어서 아래 행을 실행한다.
$ ./bin/flume node_nowatch -n collector01

tail과 tailDir 소스 설정법

규칙 : tail("filename"[,startFromEnd=false]{.delim="regex", delimMode="exclude|prev|next"})
사용예 : tail("파일명")

규칙 : 약("dirname"[, fileregex=".*"[, startFromEnd=(false[, recurseDepth=0]]]{,delim="regex", delimMode="exclude|prev|next"})
사용예 : tailDir("디렉터리명)

실습을 위한 flume master 설정
http://localhost:35871/
위 url로 들어가서 config에서 설정하는 방법과 쉘 커멘드에서 설정하는 방법이 있다.

아래는 쉘 커멘드로 설정하는 방법이다.

flume 쉘 커멘드 실행

새로운 터미널을 연다.
$./flume shell -c localhost

exec config agent 'tail("/usr/lib/flume/test.txt")' 'agentSink("localhost",35853)'

exec config collector 'collectorSource(35853)' 'collectorSink("file:///usr/local/flume/collected", "test")'

exec config agent 'tailDir("/usr/local/hadoop-1.0.4/logs/")' 'agentSink("localhost",35853)'

exec config collector 'collectorSource(35853)' 'collectorSink("file:///usr/local/flume/collected", hadoop_log")'

flume master에서 결과 확인
http://localhost:35871/

colleced 폴더에 로그파일이 수집된 모습

수집된 파일의 내용