빅데이터 수집

빅데이터

빅데이터 수집

긔 2020. 6. 25. 18:36

3. 빅데이터 수집

빅데이터 수집 개요
빅데이터 수집에 활용되는 기술
수집 파일럿 실행 1단계 - 수집 아키텍처
수집 파일럿 실행 2단계 - 수집 환경 구성
수집 파일럿 실행 3단계 - 플럼 수집 기능 구현
수집 파일럿 실행 4단계 - 카프카 수집 기능 구현
수집 파일럿 실행 5단계 - 수집 기능 테스트

빅데이터 수집 개요

빅데이터 시스템 구축은 수집부터 시작
수집이 전체 공정의 절반 이상 차지

수집 범위 P106 그림 3-2 참고

조직 전체 시스템
외부 시스템(SNS, 포털, 정부기관 등)

빅데이터 프로세싱

과거

수집 / 적재 후 맵리듀스 기반의 주기적인 배치성 분석

현재

수집과 동시에 분석
ESP(Event Stream Proccessing)
파일럿 : 운저자 상태 정보 실시간 수집하고 다양한 운행패턴 이벤트 감지 기능 구현

빅데이터 수집에 활용할 기술

플럼(Flume)

http://flume.apache.org
빅데이터 수집시 다양한 수집 요구 사항 해결
Source => Interceptor => Channel => Sink => 적재
Source : 다양한 원천 시스템 데이터를 수집하기 위한 컴포넌트 제공, 데이터로드
Interceptor: Source 와 Channel사이 데이터 필터링
Channel: 데이터 임시 저장, 데이터를 버퍼링 하는 컴포넌트로 메모리, 파일, 데이터베이스를 저장소로 활용
Sink : 최종 목적지(HDFS, Hive,..)에 저장하는 기능

플럼 구성

스마트카 상태 정보 CarLogMain 대용량

3초간격, 100M/day(대용량) : 플럼 => 하둡

스마트카 운전자 운행 정보DriverLogMain 실시간

실시간, 400kb/1초 : 플럼 => 카프카

카프카(Kafka) - MOM(Message Oriented Middleware)

대규모 메세지성 데이터를 비동기 방식으로 중계하는 역할
실시간 빠르게 발생하는 데이터 수집 시 이를 최종 목적지에 전달하기 전 중간에서 안정적인 버퍼링 처리 필요
중간 저장소가 완충 역할을 함으로써 안정적인 수집 아키텍처 구성
대규모 트랜잭션 => 중간버퍼링(카프카) => 목적지
주키퍼와 연동 필요
Broker
- 카프카 서비스의 인스터스, 다수의 Borker 클러스터로 구성
- Topic 이 생성되는 물리적 서버
Topic : Broker에서 데이터 발생/소비 처리를 위한 저장소
Producer : Borker 특정 Topic 에 전송하는 역할
Consumer: Borker 특정 Topic 에 수신하는 역할

플럼 에이전트 1

스마트카 상태 정보를 기록한 로그 파일을 일별로 수집하기 위한 배치성 플럼 에이전트
SpoolDir Source : 약속된 로그 발생 디렉토리를 모니터링하다가 정의된 로그 파일 발생 시 해당 파일의 내용을 읽어서 수집하는 기능 제공
Memory Channel: ource로 부터 수집된 데이터를 메모리 Chennel에 중간 적재
Logger Sink: Channel 로부터 읽어 들인 데이터를 플럼 표준 로그 파일로 출력

플럼 에이전트 2

스마트카 운전자의 운행 정보를 실시간으로 수집하기 위한 실시간성 플럼 에이전트
Exec-Tail Source: 로그가 쌓이고 있는 파일에 Tail 파이프라인을 이용해 실시간으로 데이터를 수집하는 기능
Memory Channel: ource로 부터 수집된 데이터를 메모리 Chennel에 중간 적재
Kafka Sink: Channel 로부터 읽어 들인 데이터를 Kafka Borker의 특적 Topic에 비동기 전송

기타

Flume Stdout: 플럼의 Logger-Sink를 통해 표준 출력 로그가 출력됨
Kafka Topic: 플럼의 Kafka-Sink는 수집된 실시간 로그를 임시 적재

수집 파일럿 실행 2단계 - 수집 환경 구성

CM을 이용해 플럼, 카프가 설치 : Server02
http://server01.hadoop.com:7180/

플럼 설치

Cluster1 > 선택 메뉴 > 서비스 추가
플럼 선택 > 호스트 선택 > Server02

플럼 설치후

설치 후 Heap 메모리 변경
Flume > 구성 > java heap 검색
50 > 100
플럼 > 선택 메뉴

카프카 설치

Cluster1 > 선택 메뉴 > 서비스 추가
Kafka 선택 > Kafka Broker > 호스트 선택 > Server02
변경 내용 검토 > 기본값 선택 후 계속

설치 중 Failde to start service 오류 발생시

Heap 메모리 변경
새로운 브라우저에 CM 접속
Kafka > 구성 > java heap 검색
50 > 256
설치페이지 > 재시도

수집 파일럿 실행 3단계 - 플럼 수집 기능 구현

2개의 에이전트 구현

SmartCar 에이전트 생성

CM > Flume > 구성 > 구성 파일 부분 찾기
CM 에서 제공하는 플럼 기본 에이전트
Agent 이름 : tier1 => SmartCar_Agent
CH03-ex-1

플럼에이전트에서 사용할 Source, Channel, Sink 변수 정의

SmartCar_Agent.sources  = SmartCarInfo_SpoolSource
SmartCar_Agent.channels = SmartCarInfo_Channel
SmartCar_Agent.sinks    = SmartCarInfo_LoggerSink

spooldir은 특정 디렉터리를 모니터링 Source.type = spooldir

새로은 파일이 생성되면 이벤트를 감지해 batchSize 값 만큼 읽어서 Channel로 전송

SmartCar_Agent.sources.SmartCarInfo_SpoolSource.type = spooldir
SmartCar_Agent.sources.SmartCarInfo_SpoolSource.spoolDir = /home/pilot-pjt/working/car-batch-log
SmartCar_Agent.sources.SmartCarInfo_SpoolSource.deletePolicy = immediate
SmartCar_Agent.sources.SmartCarInfo_SpoolSource.batchSize = 1000

중간 적재할 채널 종류 설정 Channel.type = memory

성능은 memory, 안정성은 file

SmartCar_Agent.channels.SmartCarInfo_Channel.type = memory
SmartCar_Agent.channels.SmartCarInfo_Channel.capacity  = 100000
SmartCar_Agent.channels.SmartCarInfo_Channel.transactionCapacity  = 10000

에이전트테스트 및 디버깅용 최종 목적지 Sink.type = logger

플럼 표준 출력 로그 파일
/var/log/flume-ng/flume-cmf-flume-AGENT-server02.hadoop.com.log

SmartCar_Agent.sinks.SmartCarInfo_LoggerSink.type = logger

interceptors 추가

Source와 Channel 중간에서 데이터 가공하는 역할
플럼 Source에서 유입되는 데이터 중 일부를 수정/추가/가공/정제 등
플럼 데이터 전송 단위 : Event = Header + Body
interceptors는 Header 특정값 추가, Body 데이터 가공
interceptors = filterInterceptor : 변수 선언
type = regex_filter: 정규 표현식을 이용해서 필터링
regex = ^\d{14} : 14자리 날짜 형식으로 시작하는 데이터
excludeEvents = false : true 이면 반대로 제외된 값 수집

SmartCar_Agent.sources.SmartCarInfo_SpoolSource.interceptors = filterInterceptor

SmartCar_Agent.sources.SmartCarInfo_SpoolSource.interceptors.filterInterceptor.type = regex_filter
SmartCar_Agent.sources.SmartCarInfo_SpoolSource.interceptors.filterInterceptor.regex = ^\\d{14}
SmartCar_Agent.sources.SmartCarInfo_SpoolSource.interceptors.filterInterceptor.excludeEvents = false

Source와 Channel Sink 연결

SmartCar_Agent.sources.SmartCarInfo_SpoolSource.channels = SmartCarInfo_Channel
SmartCar_Agent.sinks.SmartCarInfo_LoggerSink.channel = SmartCarInfo_Channel

DriverCarInfo 에이전트 생성

변수 추가

SmartCar_Agent.sources  = SmartCarInfo_SpoolSource DriverCarInfo_TailSource
SmartCar_Agent.channels = SmartCarInfo_Channel DriverCarInfo_Channel
SmartCar_Agent.sinks    = SmartCarInfo_LoggerSink DriverCarInfo_KafkaSink

Source.type = exec

외부 수행 명령 결과를 플럼 Event로 가져와 수집
tail -F /home/pilot-pjt/working/driver-realtime-log/SmartCarDriverInfo.log

SmartCar_Agent.sources.DriverCarInfo_TailSource.type = exec
SmartCar_Agent.sources.DriverCarInfo_TailSource.command = tail -F /home/pilot-pjt/working/driver-realtime-log/SmartCarDriverInfo.log
SmartCar_Agent.sources.DriverCarInfo_TailSource.restart = true
SmartCar_Agent.sources.DriverCarInfo_TailSource.batchSize = 1000

interceptors 추가

SmartCar_Agent.sources.DriverCarInfo_TailSource.interceptors = filterInterceptor2

SmartCar_Agent.sources.DriverCarInfo_TailSource.interceptors.filterInterceptor2.type = regex_filter
SmartCar_Agent.sources.DriverCarInfo_TailSource.interceptors.filterInterceptor2.regex = ^\\d{14}
SmartCar_Agent.sources.DriverCarInfo_TailSource.interceptors.filterInterceptor2.excludeEvents = false

ink.type = org.apache.flume.sink.kafka.KafkaSink

카프카(브로커) 설치 서버 : server02.hadoop.com:9092

SmartCar_Agent.sinks.DriverCarInfo_KafkaSink.type = org.apache.flume.sink.kafka.KafkaSink
SmartCar_Agent.sinks.DriverCarInfo_KafkaSink.topic = SmartCar-Topic
SmartCar_Agent.sinks.DriverCarInfo_KafkaSink.brokerList = server02.hadoop.com:9092
SmartCar_Agent.sinks.DriverCarInfo_KafkaSink.requiredAcks = 1
SmartCar_Agent.sinks.DriverCarInfo_KafkaSink.batchSize = 1000

Channel.type = memory

SmartCar_Agent.channels.DriverCarInfo_Channel.type = memory
SmartCar_Agent.channels.DriverCarInfo_Channel.capacity= 100000
SmartCar_Agent.channels.DriverCarInfo_Channel.transactionCapacity = 10000

Source와 Channel Sink 연결

SmartCar_Agent.sources.DriverCarInfo_TailSource.channels = DriverCarInfo_Channel
SmartCar_Agent.sinks.DriverCarInfo_KafkaSink.channel = DriverCarInfo_Channel

수집 파일럿 실행 4단계 - 카프카 기능 구현

카프카 명령어를 이용 카프카 Broker 안에서 사용할 Topic 생성
Producer 명령어를 통해 데이터 전송
Consumer 명령어로 수신

카프카 Topic 생성

Server02 SSH 접속
kafka-topics --create --zookeeper server02.hadoop.com:2181 --replication-factor 1 --partitions 1 --topic SmartCar-Topic
결과: Created topic SmartCar-Topic
이미 만들어져 있는 경우
kafka-topics --create --zookeeper server02.hadoop.com:2181 --topic SmartCar-Topic 입력해서 삭제
replication-factor 1: 다중 복제 수
partitions 1: 분산 저장 수

카프카 Topic 생성

kafka-topics --delete --zookeeper server02.hadoop.com:2181 --topic SmartCar-Topic
결과: Topic SmartCar-Topic is marked for deletion.

카프카 Producer 사용

kafka-console-producer --broker-list server02.hadoop.com:9092 --topic SmartCar-Topic

카프카 Consumer 사용

새로운 Server02 SSH 접속
kafka-console-consumer --zookeeper server02.hadoop.com:2181 --topic SmartCar-Topic --from-beggining => zookeeper is not a recognized option <오류>
kafka-console-consumer --bootstrap-server server02.hadoop.com:9092 --topic SmartCar-Topic

카프카 Producer, consumer 사용

송수신 테스트
수신 안될시 카프카 > 구성 > offsets.topic.replication.factor = 1로 변경

수집 파일럿 실행 5단계 - 수집 기능 테스트

Sever02 SSH 접속 시뮬레이터 위치로 이동
cd /home/pilot-pjt/working
3대 스마트카 상태 정보 수집

운전정보 시뮬레이터 File 실행 DriverLogMain 날짜 차량수 &(backg)

java -cp bigdata.smartcar.loggen-1.0.jar com.wikibook.bigdata.smartcar.loggen.DriverLogMain 20190101 3 &

Car정보 시뮬레이터 File 실행 CarLogMain 날짜 차량수 &(backg)

java -cp bigdata.smartcar.loggen-1.0.jar com.wikibook.bigdata.smartcar.loggen.CarLogMain 20190101 3 &

시뮬레이터 정상 작동 확인

ls -l /home/pilot-pjt/working/SmartCar/
SmartCarStatusInfo_20190101.txt 생성 확인
ls -l /home/pilot-pjt/working/driver-realtime-log/
SmartCarDriverInfo.log 생성 확인
실시간 확인 : tail -f /home/pilot-pjt/working/driver-realtime-log/SmartCarDriverInfo.log
로그파일 Spooldir 경로로 이동
mv /home/pilot-pjt/working/SmartCar/SmartCarStatusInfo_20190101.txt /home/pilot-pjt/working/car-batch-log/

수집 기능 점검

스마트카 상태 정보 로그파일 => 표준 출력로그 tail 로 확인

tail -f /var/log/flume-ng/flume-cmf-flume-AGENT-server02.hadoop.com.log
Event, Header, Body

실시간 운전정보 DriverCarInfo 수집 확인

kafka-console-consumer --bootstrap-server server02.hadoop.com:9092 --topic SmartCar-Topic

시뮬레이터 종료

ps -ef |grep smartcar.log
kill -9 pid

Tip _ 파일럿 환경 로그 확인

Hadoop 에코시스템 서버들 위치 : /var/log/각서버(cloudera,Hadoop, Oozie)등
Redis: /var/log/redis_6379.log
Storm : /home/pilot-pjt/storm/logs/
Zeppelin: /home/pilot-pjt/zeppelin-x.x.x-bin-all/logs