[2024.02.20] chapter3. 빅데이터의 분산 처리 (1) - MY

[chominyeong]

3-1 대규모 분산 처리의 프레임워크

📁 구조화 데이터(structured data)

• 스키마가 명확하게 정의된 데이터

스키마(schema) : SQL로 데이터를 집계할 때, 테이블의 칼럼 명 / 데이터형 / 테이블 간의 관계 등을 의미

📁 비구조화 데이터(unstructured data)

• 스키마가 없는 데이터
• 텍스트, 이미지, 동영상 데이터
• 비구조화 데이터는 SQL로 집계 불가능
• 비구조화 데이터를 분산 스토리지에 저장하고, 그것을 분산 시스템에서 처리하는 것이 데이터 레이크이다. → 구조화된 데이터로 변환

📁 반구조화 데이터 = 스키마리스 데이터(schemaless data)

• 서식(csv, json, xml)은 정해져 있지만, 컬럼수, 데이터형 등 스키마가 정의되지 않은 데이터
• 최근에는 JSON 형식을 이용하는 경우가 많음
  • JSON은 그대로 저장하고 거기서 필요한 필드만 추출 가능, 원래 데이터 보존

👉 데이터 구조화의 파이프라인

1. 데이터 소스에서 수집된 비구조화 데이터, 스키마리스 데이터는 처음에 분산 스토리지에 저장됨.
   a. 웹 서버의 로그 파일, 업무용 데이터베이스에서 추출한 마스터 데이터 등
2. 분산 스토리지에 수집된 데이터는 바로 SQL로 집계 불가 → 스키마를 명확하게 한 테이블 형식의 구조화 데이터로 변환
3. 변환된 구조화 데이터는 열 지향 스토리지로 저장.
   a. MPP 데이터베이스로 전송 or
   b. Hadoop 상에서 열 지향 스토리지 형식으로 변환
4. 구조화 데이터 중 시간에 따라 증가하는 데이터를 팩트 테이블, 그에 따른 부속 테이블을 디멘전 테이블로 취급

👉 열 지향 스토리지의 작성

• MPP - 제품에 따라 스토리지의 형식 고정 → 사용자가 몰라도 괜찮음
• Hadoop - 사용자가 직접 열 지향 스토리지의 형식 선택 & 자신이 좋아하는 쿼리 엔진에서 집계 가능
• Hadoop에서 사용하는 열 지향 스토리지의 종류
  • Apache ORC : 구조화 데이터를 위한 열 지향 스토리지(this book)
  • Apache Parquet : 스키마리스에 가까운 데이터 구조

정리

비구조화 데이터 --- (분산 처리 프레임워크) ---> 열 지향 스토리지로 변환

🐘 Hadoop

---

Hadoop은 분산 시스템을 구성하는 다.수.의 소프트웨어로 이루어진 집합체다.

Hadoop의 구성요소

• 분산 파일 시스템(distributed file system) : HDFS(Hadoop Distributed File System)
• 리소스 관리자(resource manage) : YARN(Yet Another Resource Negotiator)
• 분산 데이터 처리(distributed data processing) : MapReduce

• 그 외의 프로젝트는 Hadoop 본체와는 독립적으로 개발되어 Hadoop을 잉요한 분산 애플리케이션으로 동작한다.
• 모든 분산 시스템이 Hadoop에 의존하는 것은 아님
  ex. 분산 파일 시스템 - HDFS, 리소스 관리자 - Mesos, 분산 데이터 처리 - Spark 사용 가능

💾 HDFS

하둡에서 제공하는 분산 파일 시스템

• 네트워크로 연결된 여러 머신의 스토리지를 관리하는 파일 시스템, 다수의 컴퓨터에 파일을 복사하여 중복성을 높인다.

💾 YARN

리소스 관리자

• CPU나 메모리 등의 계산 리소스 관리
  • CPU : 컴퓨터의 뇌에 해당. 0과 1로 계산을 어떻게 어떤 순서로 할지 담당
  • Memory : CPU가 프로그램을 실행하거나 데이터를 참조하기 위해 필요한 정보들을 저장해두는 공간
• YARN은 CPU 코어와 메모리를 컨테이너(container) 단위로 관리한다.
• Hadoop에서 분산 애플리케이션을 실행하면 YARN이 클러스터 전체의 부하를 보고 비어 있는 호스트부터 컨테이너를 할당한다.
• 리소스 관리자는 어느 애플리케이션에 얼마만큼의 리소스를 할당할 지 관리함으로써 모든 애플리케이션이 차질없이 실행되도록 제어한다.

YARN container

컨테이너라고 하면 가상화 기술 docker를 떠올릴 수도 있다. YARN의 컨테이너는 OS 수준의 가상화 기술이 아니라 어떤 호스트에서 어떤 프로세스를 실행시킬 것인지 결정하는 애플리케이션 수준의 기술이다.

💾 MapReduce

분산 데이터 처리

• YARN 상에서 동작하는 분산 애플리케이션 중 하나, 분산 시스템에서 데이터 처리를 실행하는 데 사용
• 비구조화 데이터를 가공하는 데 적합
• 대량의 데이터를 배치 처리하기에 적합(작은 프로그램은 오버헤드가 매우 큼) → Hive도 마찬가지

💾 Hive

Q.그러면 입력 데이터 → 비구조화 테이블 / 출력 데이터 → 구조화 테이블(팩트, 디멘전) ??

쿼리 엔진

• SQL에 의한 데이터 집계가 목적이면, 이를 위해 설계된 쿼리 엔진을 사용 → 'Apache Hive'
• 쿼리를 자동으로 MapReduce 프로그램으로 변환하는 소프트웨어

💾 Hive on Tez

• Apache Tez : Hive 가속화
• Tez는 MapReduce의 단점을 해소해 고속화를 실현함으로써 MapReduce를 대체할 목적으로 개발됨
• MapReduce → 1회의 MapReduce 스트에지가 끝날 때까지 다음 처리 진행 불가
• Tez → 스테이지의 종료를 기다리지 않고 처리가 끝난 데이터는 차례대로 후속 처리에 전달
  • 현재의 Hive ; Hive on Tez
  • 예전의 Hive ; Hive on MR

💾 대화형 쿼리 엔진

쿼리 엔진

• 대화형 쿼리 실행만 전문으로 하는 쿼리 엔진
• Apache Impala, Presto
• 순간 최대 속도를 높이기 위해 모든 오버헤드가 제거되어 사용할 수 있는 리소스를 최대한 활용하여 쿼리 실행
• 대화형 쿼리 엔진은 MPP와 비교해도 손색없는 응답 시간 실현

• Hadoop에서는 다수의 쿼리 엔진이 개발되고 있음 → 총칭해 'SQL-on-Hadoop'이라고 부른다.
  
  
  
  

💫 Spark

---

인 메모리 고속 데이터 처리

• 대량의 메모리를 활용하여 고속화 실현
  • MapReduce → 처리해야 할 데이터 양에 비해 훨씬 적은 메모리만 사용 가능, MapReduce는 그 처리의 대부분을 데이터 읽고 쓰기에 사용, Tez도 마찬가지
• Spark는 '가능한 한 많은 데이터를 메모리상에 올린 상태로 두어 디스크에는 아무것도 기록하지 않는다'
• 컴퓨터가 비정상 종료하면 중간까지 처리한 중간 데이터는 사라지지만, 처리를 다시 시도해서 잃어버린 중간 데이터를 다시 만든다.

→ Spark는 Hadoop이 아닌, MapReduce를 대체한다.

• Spark는 HDFS와 YARN과 같이 사용 가능
• 실행은 자바 런타임 필요
• Spark 상에서 실행되는 데이터 처리는 자바, 스칼라, 파이썬, R 사용 가능
• Spark SQL(SQL로 쿼리 실행), Spark Streaming(스트림 처리 수행) 포함

[chominyeong]

3-2 쿼리 엔진

📌 파이프라인

1. 비구조화/스키마리스 데이터 –[Hive]→ 데이터 구조화 및 열지향 스토리지로 저장
2. 구조화 데이터 –[Presto]→ 비정규화 테이블로 데이터 마트에 보냄

1. 데이터 구조화

# Hive 기동
$ hive

# 외부 테이블 'access_los_csv'를 정의
hive> CREATE EXTERNAL TABLE access_log_csv(
    >    time string, request string, status int, bytes int
    >    )
    
    > # csv 형식임을 지정
    > ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.OpenCSVSerde'
    
    > # 경로를 지정(디렉토리 내의 모든 파일이 읽혀짐)
    > STORED AS TEXTFILE LOCATION '/var/log/access_log/'
    
    > # csv의 헤더행 스킵
    > TBLPROPERTIES ('skip.header.line.count'='1');
OK
Time taken: 1.938 seconds

• 외부 테이블 : Hive의 외부에 있는 특정 파일을 참고해 마치 거기에 테이블이 존재하는 것처럼 읽어 들이기 위해 지정
• 위의 코드에서는 'access_log_csv'라는 테이블명을 참고해 데이터를 추출 → 텍스트 파일이 로드되고 구조화 데이터로의 변환이 이루어짐

# 상태별로 레코드 수를 센다.
hive> SELECT status, count(*) cnt
    > FROM access_log_csv GROUP BY status LIMIT 2;  #2개의 행 출력
...
OK
200    1701534
302    46573
Time taken: 8.664 seconds, Fetched: 2 row(s)

• SQL-on-Hadoop의 쿼리 엔진은 데이터를 내부로 가져오지 않아도 텍스트 파일을 그대로 집계할 수 있다.
• 외부 테이블로 지정한 경로에 포함된 모든 CSV 파일이 로드되고 집계된다.
• 데이터 수를 세는데=집계하는 데 8초 이상 걸린다. → 열 지향 스토리지로 변환해줄 필요가 있다.
  
  
  

1.5. 열지향 스토리지로 변환

데이터 집계의 고속화를 위해, (배치형 쿼리 엔진용)

# ORC 형식의 테이블 'access_log_orc'로 변환
hive> CREATE TABLE access_log_orc STORED AS ORC AS
    > SELECT cast(time AS timestamp) time,
    >        request,
    >        status,
    >        cast(bytes AS bigint) bytes
    > FROM access_log_csv;
OK
Time taken: 15.993 seconds


# ORC 형식의 테이블을 집계함
hive> SELECT statux, count(*) cnt
    > FROM access_log_orc GROUP BY status LIMIT 2;
...
OK
200        1701534
302       46573
Time taken: 1.567 seconds, Fetched: 2 row(s)

ORC : Optimized Row Columnar 컬럼 기반으로 데이터 저장=열 지향

• ORC 형식으로 변환할 때에는 시간이 다소 걸리지만, 변환 후 테이블의 집계는 1.5초로 1/10 단축되었다.
• 열 지향 스토리지로 변환함으로써 집계에 시간이 고속화된다. → Hive와 같은 배치형 쿼리 엔진에서 적합
  
  
  

2. 비정규화 테이블 만들기

비정규화 테이블 만들기
├── Hive(배치형 쿼리 엔진)
│     ├── 서브 쿼리 안에서 레코드 수를 줄이는 방법
│     └── 데이터의 편향을 방지하는 방법
└── Presto(대화형 쿼리 엔진)

Hive와 Presto의 차이

• Hive
  • 시간이 걸리는 배치 처리는 Hive를 사용한다.
  • 비정규화 테이블이 수억 레코드나 되면, 이것을 데이터 마트에 보내는 것만으로도 상당한 시간이 걸리므로 배치형 시스템을 사용하는 것이 리소스 이용 효율을 높인다.
• Presto
  • 작은 쿼리를 여러 번 실행할 때 사용한다.

2-a. Hive로 비정규화 테이블 만들기

👉 서브 쿼리 안에서 레코드 수 줄이기

• Hive는 데이터베이스가 아닌 데이터 처리를 위한 배치 처리 구조임 → 데이터의 양을 의식하면서 쿼리를 작성해야 한다.

① 비효율적인 쿼리
SELECT ...
FROM access_log a
JOIN users b ON b.id = a.user_id
WHERE b.created_at = '2017-01-01'

• 팩트 테이블("access_log")와 디멘전 테이블("user")을 결합하고 WHERE로 조건을 부여하는 쿼리
• 모든 데이터를 읽어들인 후 WHERE 조건 검색
• 대량의 중간 데이터가 생성되고, 대부분은 그냥 버림 → 낭비

② 보다 효율적인 쿼리
SELECT ...
FROM (
    SELECT * access_log
    WHERE time >= TIMESTAMP '2017-01-01 00:00:00'
) a
JOIN users b ON b.id = a.user_id
WHERE b.created_at = '2017-01-01'

• 서브 쿼리 안에 팩트 테이블을 작게 생성한다.

👉 데이터 편향 피하기

분산 시스템의 성능 발휘를 위해

• 중복이 없는 값을 추출(distinct count)하기 위해서는 데이터를 한 곳에 모아야 한다. → 분산처리가 어려워짐

1. 일별 고유 유저 수의 추이 확인

① 비효율적인 쿼리
SELECT date, count(distinct user_id) users   #distinct count는 분산되지 않음
FROM access_log GROUP BY date

• distinct count는 분산되지 않아도 GROUP BY에 의한 그룹화는 분산 처리된다.
  ex. 30일 동안의 데이터 → 30으로 분할, 충분히 고속으로 실행(but 하루 데이터 양이 거의 균등하다는 조건 하에 이루어짐. 데이터에 편차가 있다면 문제 발생)

2. 웹페이지당 고유 방문자 수 확인

② 보다 효율적인 쿼리
SELECT date, count(*) useres
FROM (
    SELECT DISTINCT date, user_id FROM access_log    #애초에 중복을 없앰
) t
GROUP BY date

• 웹페이지별로 조회수에는 차이가 존재 → 편차 발생
• SELECT DISTINCT 사용

2-b. Presto로 비정규화 테이블 만들기

👍 플러그인 가능한 스토리지

하나의 쿼리 안에서 여러 데이터 소스에 연결 가능

• Presto는 데이터 소스에서 직접 데이터를 읽어 들인다.
  • 일반적인 MPP DB → 데이터를 로드하지 않으면 집계 시작 불가
• Presto는 Hive 메타 스토어에 등록된 테이블을 가져올 수 있다.
• Presto 는 열 지향 데이터에서 뛰어난 성능을 보인다. → 확장성이 높은 분산 스토리지에 배치해서 성능 발휘
• Presto 클러스터를 분산 스토리지와 네트워크의 가까운 곳에 설치한 후 가능한 한 고속 네트워크에 연결하면 데이터의 로딩 속도를 높일 수 있다.

👍 CPU 처리의 최적화

• Presto는 쿼리를 분석하여 최적의 실행 계획 생성 → 바이트 코드로 변환 → 이 바이트 코드가 Presto의 워커 노드에 배포됨 → 런타임 시스템에 의해 기계 코드로 컴파일 된다
• 코드는 멀티 스레드화되어 단일 머신에서 수백 태스크 or 병렬로 실행된다. + 열지향 스토리지에서의 읽기도 병렬화되어 데이터가 도달할 때마다 처리가 진행됨
  • →Presto의 CPU 이용 효율이 높기 때문에, 메모리와 CPU 리소스만 충분하면 데이터의 읽기 속도가 쿼리의 실행 시간을 결정하게 된다.
• Presto 쿼리는 실행되면 중간에 끼어들 수 없음 → 너무 큰 쿼리 실행하면 안됨
• 대부분의 쿼리는 단시간에 종료해 리소스가 해제됨

👍 인 메모리 처리에 의한 고속화

• Presto는 쿼리의 실행 과정에서 디스크에 쓰기를 하지 않는다.
  • Presto는 모든 데이터 처리를 메모리상에서 실시하고 메모리가 부족하면 여유가 생길 때까지 기다리거나 오류로 실패한다.
  • 메모리 할당을 늘리거나 쿼리를 다시 작성해 메모리 소비를 줄여야 한다.
• 데이터 양이 많아도 메모리 소비는 늘어나지 않는다.
  • GROUP BY는 단순 반복이므로 메모리 소비량은 고정이다.
• 메모리상에서 할 수 있는 것은 메모리에서, 디스크가 있어야 하는 task는 Hive에서 실행

👍 분산 결합과 브로드캐스트 결합

• 2개의 팩트 테이블을 결합하는 경우는 분산 결합(distribute join) 실시
• 매우 많은 조인 키를 메모리상에 계속 유지해야 함
• 같은 키를 갖는 데이터는 동일한 노드에

• 한쪽 테이블이 작은 경우에는 브로드캐스트 결합(broadcast join) 실시
• 결합하는 테이블의 모든 데이터가 각 노드에 복사됨
  • 팩트 테이블은 냅두고, 디멘전 테이블은 여러 개 복사
• Presto에서 브로드캐스트 결합을 하려면 분산 결합은 명시적으로 무효화해야 한다. + 쿼리 안의 SELECT 문으로 먼저 팩트 테이블을 지정하여 거기에 디멘전을 결합해야 한다.

👍 열 지향 스토리지 집계

# Presto 기동(Hive 메타 스토어 이용)
$ presto --catalog hive --schema default

# ORC 형식의 테이블 집계
presto:default> SELECT status, count(*) cnt
             -> FROM access_log_orc GROUP BY status LIMIT 2;
status |  cnt
-------+--------
200    | 1701534
302    |   46573
(2 rows)

Query 20170520_152030_00005_u8m9e, FINISHED, 1 node
Splits: 50 total, 50 done (100.00%)
0:00 [1.89M rows, 7.96MB] [4.85M rows/s, 20.4MB/s]

🔩

비구조화 데이터 ---(데이터 구조화)---> 구조화 데이터(열지향 스토리지) ---(데이터 집약)---> 비정규화 테이블(데이터 마트)