빅데이터를 지탱하는 기술 (5.2~5.3 배치/스트리밍 형의 데이터 플로우)

[5부 빅데이터의 파이프라인]

5-2 배치 형의 데이터 플로우
1) MapReduce의 시대는 끝났다 ― 데이터 플로우와 워크플로
2) MapReduce를 대신할 새로운 프레임워크 ― DAG에 의한 내부 표현
3) 데이터 플로우와 워크플로를 조합하기
4) 데이터 플로우와 SQL을 나누어 사용하기 ― 데이터 웨어하우스의 파이프라인과 데이터 마트의 파이프라인

5-3 스트리밍 형의 데이터 플로우
1) 배치 처리와 스트림 처리로 경로 나누기
2) 배치 처리와 스트림 처리 통합하기
3) 스트림 처리의 결과를 배치 처리로 치환하기 ― 스트림 처리의 두 가지 문제에 대한 대처
4) 아웃 오브 오더의 데이터 처리

5-4 정리

 

---

 

5부 빅데이터의 파이프라인

 

5-2 배치 형의 데이터 플로우

 
1) MapReduce의 시대는 끝났다 ― 데이터 플로우와 워크플로

- 데이터 플로우
=> 분산 스토리지로의 데이터 전송이 완료되면 분산 시스템의 프레임워크를 사용해 SQL만이 아닌 프로그래밍 언어를 사용해 데이터 파이프라인을 작성
=> 기술적 발전에 따라 다단계의 데이터 처리를 그대로 분산 시스템의 내부에서 실행할 수 있게 됨
* 이전에는 MapReduce를 사용한 데이터 처리에서는 MapReduce 프로그램을 워크플로의 태스크로 등록해 다단계의 복잡한 데이터 처리를 할 수 있었음
=> 데이터 플로우를 위한 프레임워크
1) Google Cloud Dataflow (Google)
2) Apache Spark (The Apache Software Foundation)
3) Apache Flink (The Apache Software Foundation)

- MapReduce의 구조
=> MapReduce 실행단계
1) 집계 대상이 되는 원래의 데이터에 대해 일정 크기로 나눈 작은 데이터인 split을 만듦
2) [Map] split을 읽어 단어별로 그중에 포함된 단어를 카운트
* 각각의 처리는 독립적이므로 다수의 컴퓨터에 분산할 수 있지만, 분산 처리의 결과는 마지막에 집계해야 함
3) [Reduce] 단어별로 그 수의 합계를 구함

=> MapReduce?
- Map과 Reduce를 반복하면서 목적하는 결과를 얻을 때까지 계속해서 데이터를 변환해 나가는 구조
* Map과 Reduce라는 하나의 사이클이 끝나지 않으면 다음 처리로 이동하지 않음

=> MapReduce의 한계
- 하나의 사이클에서 다음 사이클로 이동할 때까지의 대기 시간이 적지 않게 발생함
- 따라서 특히 '애드 혹 데이터 분석'에서 요구되는 <지연이 적은 집계>는 MapReduce로 실현하는 것이 어려움

 

---

 
2) MapReduce를 대신할 새로운 프레임워크 ― DAG에 의한 내부 표현 

- MapReduce를 대체하는 프레임워크 (과거의 빅데이터 기술로 간주되는 MapReduce)
1) 구글은 차세대 기술로 'MillWheel' 프레임워크를 개발해 클라우드 서비스인 Google Cloud Dataflow 내부에서 이용
2) Hadoop에서 개발한 Tez
3) Spark

- DAG (Directed Acyclic Graph, 방향성 비순환 그래프)
=> DAG는 수학/컴퓨터 알고리즘에서 사용되는 데이터 모델의 하나임 (어떤 기술이 아님)
=> DAG의 특성
1) 노드와 노드가 화살표로 연결됨 (방향성)
2) 화살표를 아무리 따라가도 동일 노드로는 되돌아오지 않음 (비순환)

- MapReduce에서의 DAG와 데이터플로우에서의 DAG 차이점
1) MapReduce (Map과 Reduce라는 두 종류의 노드로 이루어진 간단한 DAG라고 할 수 있음)
=> 하나의 노드에서 처리가 끝나지 않으면 다음 처리로 진행할 수 없으므로 비효율적임

2) 데이터플로우에서의 DAG
=> 각 구성 노드가 모두 동시 병행으로 실행되어 처리가 끝난 데이터는 네트워크를 거쳐 차례대로 전달되어 MapReduce에서의 대기시간이 없음

- Spark에 있어서의 DAG
=> DAG에 의한 프로그래밍의 특징은 '지연 평가 (lazy evaluation)'
=> 먼저 DAG를 구축하고 그 후에 명시/암묵적으로 실행 결과를 요구함에 따라 데이터 처리를 시작함
* MapReduce처럼 Map과 Reduce를 하나씩 실행하는 것이 아닌, 먼저 데이터 파이프라인 전체를 DAG로 조립하고 나서 실행에 옮김으로써 내부 스케줄러가 분산 시스템에 효과적인 실행 계획을 세워주는 것이 데이터 플로우의 장점임

 

• [배치 처리] Spark에서 데이터 처리를 하는 파이썬 스크립트 예시

# 1. 파일로부터 데이터를 읽어 들임 (textFile())
lines = sc.textFile("sample.txt")

# 2. 파일의 각 행을 단어로 분해하여 결합 (flatMap())
words = lines.flatMap(lambda line: line.split())

# 3. 단어마다의 카운터(map())의 
# 4. 합계를 계산해(reduceByKey()) 
# 5. 파일에 출력(saveAsTextFile())
words.map(lambda word: (word, 1)) \
	 .reduceByKey(lambda a, b: a + b) \
     .saveAsTextFile("word_counts") # 여기에서 실행 개시

 

---

 

3) 데이터 플로우와 워크플로를 조합하기 

- 데이터 플로우 프로그래밍을 통해 데이터의 입출력을 하나의 DAG로 기술
=> 데이터 플로우의 프로그램도 다시 워크플로의 일부로서 실행되는 하나의 태스크로 고려할 수 있음
* 태스크의 정기 실행이나 실패한 태스크의 기록/복구는 데이터 플로우에서 할 수 없음
=> 분산시스템 안에서만 실행되는 데이터 처리라면 하나의 데이터 플로우로 기술할 수 있음
=> 분산시스템의 외부와 데이터를 주고 받을 경우 복구를 고려해 워크플로 안에서 실행하는 것이 바람직함 

1. 데이터를 읽어들이는 플로우
=> 데이터 플로우에서 읽어 들일 데이터는 성능적으로 안정된 분산 스토리지에 배치
* 데이터 소스에 액세스하면 성능 문제를 일으키기 쉬우므로 워크플로를 이용하고 데이터 플로우에서는 분산 스토리지에 복사함으로써 안정적으로 사용
=> 외부의 데이터 소스에서 데이터를 읽어 들일 때는 벌크 형의 전송 도구로 태스크를 구현
* 데이터 플로우를 사용한다고 데이터 소스에서의 읽기 속도가 빨라진다고 단언할 수 없지만, 오류 발생에 대해 확실하게 대처해 복사를 끝내도록 하는 것이 선결 과제

2. 데이터를 써서 내보내는 플로우
=> 데이터 집계결과를 외부 시스템에 써서 내보내는 경우 어떤 오류가 발생할지 예측할 수 없으므로 '데이터를 읽어들이는 플로우'와 반대임
* 데이터 플로우 안에서 대량의 데이터를 외부에 전송하는 것을 피하는 것이 무난함
=> 데이터 플로우에서는 CSV 같이 취급하기 쉬운 형식으로 변환해 분산 스토리지에 써넣음
* 워크플로의 역할은 외부 시스템에 데이터를 전송하는 것임

 

---

 
4) 데이터 플로우와 SQL을 나누어 사용하기 ― 데이터 웨어하우스의 파이프라인과 데이터 마트의 파이프라인

1. SQL을 MPP 데이터베이스에서 실행하는 경우 
=> 데이터 웨어하우스의 파이프라인
=> 데이터 플로우의 역할: 로드되는 데이터를 만드는 것까지
* 비구조화 데이터를 가공해 CSV 파일 등을 만들어 분산 스토리지에 써넣음 -> 워크플로 통해 태스크 실행이나 SQL 쿼리 실행

2. 분산 시스템상의 쿼리 엔진에서 실행하는 경우
=> 데이터마트의 파이프라인
=> 데이터 플로우의 역할: 구조화 데이터를 만드는 부분까지
* 분산 스토리지 상의 데이터를 배치로 가공해 열 지향 스토리지 형식으로 보관 -> 워크플로 통해 SQL 실행(쿼리 엔진 사용)이나 결과를 데이터 마트에 써서 내보냄

3. 애드 훅 분석 시의 대화식 플로우
=> 애드 훅 분석의 파이프라인
* 애드 훅 분석에서는 많은 데이터 처리를 수작업으로 시행하므로 워크플로는 필요하지 않음

=> 아직 구조화되어 있지 않은 데이터를 애드 훅으로 분석할 때 데이터 플로우는 매우 유용함
* 로우 데이터에 직접 접속해 스크립트 언어를 사용해 데이터 가공/집계하여 데이터를 구조화함 -> 구조화된 데이터에 대한 집계는 고속으로 처리가 가능함

=> 이미 구조화되어 있는 데이터의 경우 쿼리 엔진을 사용해 참조하도록 함
* 커멘드라인/SQL 실행 등으로 시각화 도구, 쿼리 엔진을 직접 접속함 (ODBC, JDBC 드라이버 사용)
* 안정적인 워크플로 운용을 추구할 경우, RDB와 MPP 데이터베이스를 데이터 마트로 하는 것이 확실함 

 

---

 

5-3 스트리밍 형의 데이터 플로우

데이터 실시간 처리를 높이기 위해서는 배치 처리와는 전혀 다른 데이터 파이프라인이 필요함
=> DAG를 사용한 스트림 처리 구조

 
1) 배치 처리와 스트림 처리로 경로 나누기

- 배치 처리 중심의 데이터 파이프라인의 단점은 데이터 집계를 위한 시간이 필요하다는 점
=> 실시간성이 높은 데이터 처리 시스템은 스트림 처리를 도입
* 실시간? 이벤트 발생에서 몇 초 후에는 결과를 알 수 있는 경우

=> 실시간성이 높은 데이터 처리 시스템의 예시
1) [시스템 모니터링] 서버와 네트워크의 상태를 감시하고, 그 시간 추이를 그래프로 표시
2) [로그 관리 시스템] OS의 시스템 이벤트나 로그 파일을 검색해 비정상적인 상태인 경우 경고 생성
3) [복합 이벤트 처리 (CEP, Complex Even Processing)] 다수의 업무 시스템에서 보낸 이벤트 데이터를 자동 처리

- 배치 처리 vs 스트림 처리 (streaming processing)
=> 배치/스트림 처리는 서로 결점을 보완하는 관계

1) 배치 처리: 과거 데이터를 집계하는 경우 사용
=> 도달한 데이터를 분산 스토리지에 보관 (데이터를 작게 나눠 DAG에 넣음) -> 정기적으로 데이터 추출해 데이터 처리 (비교적 큰 단위인 1시간마다)
* 데이터가 영속적으로 보존되기 때문에 계속해서 재실행 가능
* 장기적인(대체로 1년 이상) 데이터 분석을 예상해 집계 효율이 높은 열 지향 스토리지를 구축할 수 있음

2) 스트림 처리: 실시간성 데이터를 집계하는 경우 사용
=> 데이터가 도달하는 것과 거의 동시에 처리가 시작 (끊임없이 데이터가 생성돼 DAG 안에 들어옴에 따라 처리가 진행) -> 처리 내용은 미리 정해 둘 필요가 없음
* 과거로 거슬러 올라가 재실시하는 것은 고려하지 않음
* 처리한 결과는 시계열 데이터에 적합한 데이터 스토어에 보관하거나, 기존 실시간 시스템에 전송함

 

---

 
2) 배치 처리와 스트림 처리 통합하기

- 유한(bounded) 데이터 vs 무한(unbounded) 데이터
=> 공통점? 데이터를 작게 분할해 DAG에서 실행한다는 점
=> 차이점?
1) 유한 데이터: 실행 시에 데이터양이 정해지는 것
2) 무한 데이터: 제한 없이 데이터가 보내지는 것

=> DAG를 사용한 데이터 플로우에서는 배치/스트림 처리를 동일하게 프로그래밍하는 것이 가능
* 예를 들어, 스트림 처리를 위한 DAG를 가공해 분산 스토리지 상의 과거 데이터를 읽어 들이도록 하면 '배치 처리'로도 가능

- Spark 스트리밍의 DAG
=> Spark는 원래 배치 처리를 위한 분산 시스템이나 'Spark 스트리밍' 기능의 통합을 통해 스트림 처리까지 가능한 프레임워크가 됨
* 데이터 읽고 쓰는 초기화 부분에서만 차이가 있을뿐, 데이터 처리의 중심부는 배치/스트리밍 처리가 동일

- 스트림 처리에 의한 1차 집계
=> 1초마다 통곗값만을 기록하고 싶은 경우에는 그 집계를 스트림 처리에 맡길 수도 있음
=> 데이터양이 너무 많아 모두 저장하고 싶지 않은 경우에는 데이터양을 삭감하기 위해 스트림 처리를 사용
* 불필요한 데이터가 전송되었다면 먼저 제외하여 스토리지 사용량을 줄일 수 있음
* 스트림 처리의 결과를 메시지 브로커에 다시 작성해 재이용할 수도 있음 -> 삭감된 데이터를 일반 메시지 전송처럼 배치 처리를 위해 분산 스토리지에 보관하고, 실시간 보고서 위한 시계열 DB 등에 전송

 

• [스트리밍 처리] Spark에서 데이터 처리를 하는 파이썬 스크립트 예시

# 1. 1초마다 스트림 처리
sc = SparkContext("local[2]", "NetworkWordCount")
ssc = StreamingContext(sc, 1)

# 2. TCP 포트 9999로부터 데이터를 읽음
lines = ssc.socketTextStream("localhost", 9999)

# 3. 입력의 각 행을 단어로 분리
words = lines.flatMap(lambda line: line.split())

# 4. 단어별 갯수를 콘솔에 출력
words.map(lambda word: (word, 1)) \
     .reduceByKey(lambda a, b: a + b) \
     .pprint()

# 5. 스트림 처리 시작
ssc.start()

 

---

 
3) 스트림 처리의 결과를 배치 처리로 치환하기 ― 스트림 처리의 두 가지 문제에 대한 대처

- 스트림 처리의 잠재적인 문제 2가지
1) 틀린 결과 데이터는 수정할 수 없음 (스트림 처리는 원칙적으로 새롭게 도달한 데이터를 처리할 뿐, 시간을 되돌리는 개념은 없음)
2) 집계가 종료된 이후에 도착한 데이터가 있어 스트림 처리의 결과가 부정확해질 수밖에 없음

- 해결방법?
=> [전통적인 대처 방법] 스트림 처리와 별개로 배치 처리를 실행시켜 후자의 결과가 옳다고 하는 것임
* 스트림 처리의 결과를 배치 처리의 결과가 나올 때까지 잠정 값으로 이용

=> [람다 아키텍처] 데이터 파이프라인을 3개의 레이어로 구분
* 장점?
- 실시간 뷰의 결과를 이후에 배치 뷰로 치환한다는 것
- 스트림 처리의 결과를 일시적으로만 사용하고, 배치 처리의 결과를 통해 올바른 결과를 얻을 수 있음
- 배치 처리만 안정되게 동작한다면 스트림 처리를 재실행할 필요없음

* 단점?
- 나쁜 개발 효율 (스피드/배치 레이어는 모두 똑같은 처리를 구현)

1) 배치 레이어 (batch layer)
* 모든 데이터는 반드시 '배치 레이어'에서 처리
* 과거의 데이터를 장기적인 스토리지에 축적하고 재집계
* 배치 레이어는 대규모 배치 처리를 실행할 수 있지만, 1회 처리에는 긴 시간이 걸림

2) 서빙 레이어 (serving ")
* 배치 처리 결과는 '서빙 레이어'를 통해 접근
* 응답이 빠른 DB를 설치해 집계 결과를 바로 추출하도록 함
* 서빙 레이어에서 얻어진 결과인 '배치 뷰(batch view)'는 정기적으로 업데이트되지만, 실시간 정보를 얻을 수 없음

3) 스피드 레이어 (speed ")
* 서빙 레이어에서 얻어진 결과인 '배치 뷰(batch view)'는 정기적으로 업데이트되지만, 실시간 정보를 얻을 수 없음
* 따라서 다른 경로로 스트림 처리를 하기 위해 '스피드 레이어'를 설치
* 스피드 레이어에서 얻은 결과인 '실시간 뷰(realtime view)'는 배치 뷰가 업데이트될 동안까지만 이용되고, 오래된 데이터 순으로 삭제됨 -> 배치 뷰, 실시간 뷰를 조합시키는 형태로 쿼리를 실행

=> [카파 아키텍처(kappa)] 배치/서빙 레이어를 제거하고 스피드 레이어만 남긴 람다 아키텍처를 단순화한 형태
* 단점?
- 부하가 높아짐 (하지만 클라우드 서비스 보급에 따라 자원 확보에 어려움이 없음)

* 메시지 브로커의 데이터 보관 기간을 충분히 길게 하여 문제 발생 시 메시지 배송 시간을 과거로 재설정해 출력 데이터를 덮어씀
* 배치 처리처럼 과거 데이터의 일괄 처리를 스트림 처리만으로 함

 

---

 
4) 아웃 오브 오더의 데이터 처리

- 아웃 오브 오더인 데이터 문제
=> 늦게 도달하는 메시지 (프로세스 시간과 이벤트 시간의 차이)

- 이벤트 시간
=> 데이터가 처음에 생성된 시간인 이벤트 시간으로 집계해야 올바른 결과를 얻을 수 있음
=> 데이터가 도달한 순간에 집계를 시작하므로 시간에 대해 특별한 조작을 하지 않는 한 출력은 프로세스 시간과 연관되고, 예기치 못한 혼란을 일으킬 수 있음

- 이벤트 시간 윈도윙 (event-time windowing)
=> 이벤트 시간에 의해 윈도우를 나누는 것
=> 이벤트 시간으로 보면 메시지가 배송된 데이터는 무작위 순으로 나열된 '아웃 오브 오더' 상태로 적절히 순서를 바꿔 집계 결과를 업데이트해야 함
=> 이벤트 시간 윈도윙을 위해서는 고려사항에 따라 DAG를 기술
* 과거 이벤트의 상태를 보관하면서 데이터가 도달할 때마다 해당하는 윈도우를 재집계할 필요가 있음
* 데이터를 계속 보관할 수 없으므로 일정 이상 늦게 온 데이터는 무시할 필요가 있음