데이터 엔지니어링

apache spark 은 데이터를 다루는 사람들에게는 필수적으로 알아야 하는 기술이다.

내 방식대로 정의하자면 spark 은 framework 이면서, 빅 데이터 처리 엔진이라고 할 수 있을 것 같다.

요즘엔 Snowflake 같은 SaaS 로 SQL Like 기반의 데이터 처리도 많이 하지만, 오픈소스 진영에서는 무조건 spark 이다.

아래 예전에 정리했던 기술을 다시 보면서, 블로그 포스팅을 해본다

1.스파크에 대해

• 하나의 서버에서 처리할 수 있는 수준 이상의 대용량 데이터 처리를 가능하게 해준다
• 고수준 API 를 제공한다
• 동종 시스템 중 가장 빠른 축에 속한다.

스파크 버전 규칙

• [메이저].[마이너].[유지보수]

왜 스칼라인가

• 스파크는 스칼라로 쓰여져 있다.
• 스파크는 함수형 프레임워크이며, 불변성이나 람다 정의 같은 개념에 크게 의존하고 있기 때문에 함수형 프로그래밍 언어인 스칼라가 좋다
• 자바 API 보다 훨씬 사용하기 쉽다
• JVM 과의 통신 비용이 랭기지중 가장 좋다. (객체 변환에 드는 비용정도)

2.스파크 동작 개념

스파크의 독보적 장점

• 메모리 기반 처리
• 지연 평가 ( lazy operation )

스파크 위치

• JVM 위에서 연산을 수행하는 것일 뿐, 데이터 저장 솔루션은 아니다
• 클러스터 매니저 종류 : 단독 클러스터매니저(StandAlone Cluster Manager), 아파치 메소스, 하둡 얀

스파크 컴포넌트

• 스파크 코어 - Java, Scala, Python, R 로 API 제, RDD 와 RDD API, JVM 사이에 데이터를 읽고 쓰는 I/O 제공
• 스파크 SQL - DataFrame ( 반구조화의 데이터 타입을 위한 인터페이스) 제공
• 스파크 스트리밍 등

병렬 연산 모델 : RDD

• 스파크의 드라이버 혹은 마스터 노드를 위한 프로그램을 사용자가 만들어야 한다
• RDD는 익스큐터 혹은 슬레이브 노드에 저장된다
• RDD를 구성하는 객체는 파티션이라 하며, 경우에 따라 다른 노드에서 계산될 수 있다.
• 클러스터 매니저는 애플리케이션에서 설정한 파라미터에 따라 익스큐터를 실행하고 분산해 주는 역할을 한다
• 드라이버가 RDD 데이터의 처리 계획을 결정하면 실행을 지연하고, 최종 RDD를 계산해야 하는 시점에 실행된다.( 보통 저장 장치에 써야 할 시점이나 결과 데이터를 드라이버에 보내야 할 때)

지연평가

• RDD의 연산 종류는 transformation 과 action 이 있다.
• 액션은 데이터를 드라이버로 되돌려주든지(count, collect 등) 데이터를 외부 저장 시스템에 쓰는것(CopyToHadoop) 등의 일이다.
• 액션은 스케쥴러를 시작하게 하며, RDD 트랜스포메이션 간의 종속성에 바탕을 둔 DAG 를 생성한다.
• 트랜스포메이션의 로직 에러가 발생할 때, 지연 평가 때문에, 액션이 수행된 지점에서 실패한 것으로 나타나는 경우에 유의하자
  • word count 프로그램에서 null pointer exception 이 발생한다고 가정할때, 코드가 contains를 체크하는 시점이 아니라 collect 단계에서 예외가 발생한다.

메모리 영속화 & 메모리 관리

• 맵리듀스와 비교해 스파크 성능상 이점은 반복 연산이 들어있는 사례이다.
• 스파크가 처리하는 데이터를 디스크에 기록하지 않고 익스큐터 메모리에 데이터를 로드해 놓는 것이다.
• 메모리 관리의 3가지 옵션
  1. 메모리에 직렬화되지 않은 자바 객체 - RDD 객체를 직렬화 하지 않고 그대로 저장한다. ( 직렬화 비용이 안드는 대신 메모리 공간 사용이 비 효율이다. )
  2. 메모리에 직렬화된 데이터 - RDD 객체를 네트워크 전송이 가능한 바이트 스트림으로 변환한다. ( 데이터를 읽는데 CPU 가 더 많이 사용되므로 접근방식은 더 느리지만 메모리 공간 사용 측면에서 효율적이다. 크리오(Kryo) 직렬화를 쓰면 공간 측면에서도 효과적이다.)
  3. 디스크 - 익스큐터 램에 담기에 파티션이 큰 RDD 라면 디스크에 데이터를 쓸 수 있다. ( 반복 연산시 속도 면에서 불리하지만 장애에 안전하다.)
• persist() 의 기본 옵션은 RDD를 메모리에 직렬화되지 않은 상태로 저장한다.

불변성과 RDD 인터페이스

• RDD는 정적인 타입인 데다 불변한 성격을 가지고 있어, Transformation 을 호출하는 것이 새롭게 정의한 속성들을 가진 새로운 RDD를 리턴하는 행위다.
• RDD 생성 방식
  • 기존 RDD에 Transformation 호출
  • SparkContext 객체로부터 생성
  • DataSet이나 DataFrame을 변환 ( 이것들은 SparkSession 으로 부터 만들어짐)
• SparkContext는 스파크 클러스터와 실행중인 스파크 애플리케이션 하나와의 연결을 나타낸다.
• RDD 속성을 알 수 있는 함수
  • partitions() - 분산 데이터 셋의 부분들을 구성하는 파티션 객체들의 배열을 리턴한다. getPartition()의 결괏값과 같다
  • iterator(p,parentIters) - 각각의 부모 파티션을 순회하는 반복자가 주어지면 파티션 p의 구성요소들을 계산해낸다.
  • dependencies() - 종속성 객체의 목록을 리턴한다. 스케쥴러가 현재의 RDD가 어떤식으로 다른 RDD에 종속될지 알려준다.
  • partitioner() - element 와 partition 사이에 연관되는 함수를 갖고 있는 RDD라면 스칼라의 option 타입으로 partitioner 객체를 리턴한다.
  • perferredLocations(p) - 파티션 p의 데이터 지역성에 대한 정보를 리턴한다. p가 저장된 각 노드의 정보를 문자열로 표현한 시퀀스를 리턴한다

RDD의 종류

• RDD는 정적인 타입인 데다 불변한 성격을 가지고 있어, Transformation 을 호출하는 것이 새롭게 정의한 속성들을 가진 새로운 RDD를 리턴하는 행위다.

넓은 종속성 vs 좁은 종속성

• 종속성이 넓으냐 좁으냐는 트랜스포메이션 평가에 중요한 영향을 끼치며 성능에도 크게 작용한다.
• 좁은 종속성
  • 자식 RDD 의 각 파티션이 부모 RDD의 파티션들에 대해 단순하고 한정적인 종속성을 갖는다
  • 부모가 최대 하나의 자식파티션을 갖는 경우
  • map, filter, mapPartitions 등의 연산이 이 패턴을 따른다.
• 넓은 종속성
  • 자식 RDD가 다수의 부모 RDD의 파티션들과 관계를 맺고 있는 경우
  • groupbykey, sort, reducebykey 등과 같이 Shuffle 을 일으키는 함수가 이 패턴을 따른다.
  • 셔플 비용이 가장 크다.
• map 은 파티션 간 이동이 없는 연산, coalesce 는 파티션을 합치는 연산으로 파티션 개수를 줄이는 목적의 함수이다.
• join 함수는 두개의 부모 RDD 가 어떻게 파티션되었는지에 따라 좁거나 넓은 종속성을 가질 수 있다.

스파크 잡 스케쥴링

• 잡 실행 과정
  • 스파크 프로그램 자체는 드라이버 노드에서 실행되며 일련의 명령들을 익스큐터에게 보낸다.
  • 애플리케이션들은 클러스터 매니저에 의해 스케쥴링되고, 각각 하나의 SparkContext를 가진다.
  • 스파크 애플리케이션들은 공존하는 여러 개의 잡을 차례로 실행할 수 있다.
  • Job들은 애플리케이션의 한 RDD가 호출하는 각 액션에 대응한다.
• 자원할당
  • 정적할당과 동적 할당이 가능
• 스파크 애플리케이션
  • 잡들은 드라이버 프로그램의 SparkContext에 정의되어 있다.
  • SparkContext가 생기면 스파크 애플리케이션이 구동한다.
  • SparkContext를 실행하면 드라이버와 익스큐터들이 각 작업 노드에서 구동된다.
  • 각 익스큐터는 JVM을 가지며 한 노드에 여러 개의 익스큐터가 존재할 수 있다.
  • 스파크 잡이 실행될 때 익스큐터는 RDD를 계산할 태스크 실행을 위한 슬롯을 가진다.
• 스파크 잡 해부
  • 각 액션마다 스파크 스케쥴러는 실행 그래프를 만들고 스파크 잡을 실행한다.
  • 각 잡은 최종 RDD를 만들어 내는데 필요한 데이터 변환의 각 단계를 의미하는 스테이지(Stage)들로 구성된다.
  • 각 스테이지는 각 병렬 연산의 한 단위를 의미하며 익스큐터들 위에서 실행되는 다수의 태스크(Task)들로 구성된다.
  • Spark Aplication -> 잡 -> 스테이지1 & 스테이지2 ... -> 태스크 1& 태스크 2...
  • SparkContext / SparkSession -> 액션 -> 넓은 포메이션 -> 하나의 파티션 평가를 위한 연산
• 잡
  • 스파크 실행 구성에서 가장 높은 단계
  • 하나의 잡 = 하나의 액션에 대응하며, 액션은 스파크 애플리케이션의 드라이버의 프로그램에서 호출한다.
• 스테이지
  • 잡의 부분들을 스테이지로 정의한다.
  • 넓은 트랜스포메이션에 의해 생성되는 셔플 의존성에 대응한다.
  • 하나의 스테이지는 다른 익스큐터나 드라이버와의 통신 없이 계산 가능한 태스크들의 집합으로 생각할 수 있다.
• 태스크
  • 하나의 스테이지는 여러 개의 태스크로 이루어진다.
  • 실행 계층에서 가장 작은 단위이며, 익스큐터가 태스크 실행을 위해 동적으로 할당된 여러개의 슬롯을 가진다.
  • 스테이지당 태스크의 개수는 해당 스테이지의 결과 RDD의 파티션 개수에 대응된다.
  • 익스큐터 코어 개수의 총합 = 익스큐터당 코어의 개수 X 익스큐터 개수를 쓰면 sparkConf로 부터 동시 실행의 태스크 개수를 구할 수 있다.
  • 태스크 분산 과정은 TaskScheduler 가 담당하는데, 페어 스케쥴러인지 FIFO 스케쥴러인지 등에 따라 다르다.

쿠버네티스 복습을 하면서 스토리지 관련해 요약했던 내용을 포스팅 해보겠다.

1.EBS란 ?

• Elastic Block Store (EBS)
• EC2 에서 쉽게 사용할 수 있는 영구적인 Volume (linux 가 볼륨을 디바이스로 인식하는)
• mkfs 명령어를 사용하여 mount 가능
• 다양한 유형
  • ssd 기반 - gp2
  • iops 속성이 존재. io1, io2., st1, sc1 등.
  • iops 조절이 가능한 gp3

2.PV란 ?

• Persistent Volume
• 물리적인 볼륨을 나타냄
• Cluster 단위의 resource (node 처럼)
• 관리자가 생성하거나 pvc 로 부터 동적으로 생성됨
• 예를 들어 pv 는 요리사가 만든 피자 한 판을 굽는 것이다.

3.PVC란?

• namespace 단위의 resource (pod 처럼)
• 사용할 용량, access mode, 등의 속성을 갖고 있음
• PV의 요청을 의미하는 추상 오브젝트
• 예를 들어 pvc 는 피자 한 판에서 피자 한 조각을 손님이 접시에 담아 가져오는 것이다.

4.provisioner 란?

• storageclass 의 설정과 pvc 의 설정을 읽어 aws ebs 등 물리적인 volume 과 pv obeject 생성을 담당.

5.csi Driver 란?

• Container Storage Interface Driver
• EKS 에서는 쿠버네티스에 내장된 provisioner 를 모두 삭제하고 별도의 컨트롤러 파드를 통해 동적으로 프로비저닝을 사용할 수 있도록 만들었음. 그것이 바로 csi driver

아래 실습을 간단히 해본다

먼저 스토리지 클래스를 만들고

apiVersion: storage.k8s.io/v1
kind: StorageClass
metadata:
  name: my-gp2
provisioner: kubernetes.io/aws-ebs
parameters:
  type: gp2
  fstype: ext4

PV를 만든다.

apiVersion: v1
kind: PersistentVolume
metadata:
  name: my-pv
spec:
  capacity:
    storage: 10Gi
  accessModes:
    - RewadWriteMany
  persistentVolumeReclaimPolicy: Retain
  storageClassName: my-gp2
hostPath:
    path: /mnt/data


---

apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: my-pvc-ebs
  namespace: nextcloud

spec:
  storageClassName: my-gp2
  accessModes:
    - ReadWriteOnce
  resources:
    requests:
      storage: 10Gi
  persistentVolumeReclaimPolicy: Retain

간단하게 Pod 를 만들어 마운트해보자

apiVersion: v1
kind: Pod
metadata:
  name: task-pv-pod
spec:
  nodeSelector:
    kubernetes.io/hostname: kube-01
  volumes:
    - name: task-pv-storage
      persistentVolumeClaim:
        claimName: task-pv-claim
  containers:
    - name: task-pv-container
      image: nginx
      ports:
        - containerPort: 80
          name: "http-server"
      volumeMounts:
        - mountPath: "/usr/share/nginx/html"
          name: task-pv-storage

카프카 구축 전에 예전에 정리했던 내용 기반으로 이론을 다시 정리해봤다. 

요즘에는 운영을 쉽게 하려고 MSK, 컨플루언트 카프카를 많이 사용 하는 것 같다.

아무튼 본론으로..

1.카프카란?

• 분산 이벤트 큐.
• 분산 이벤트 스트리밍 플랫폼
• 카프카 컨슈머, 프로듀서, 스트림즈, 커넥트 등 연동 API 제공
• 초당 수백만개 데이터를 처리할수 있으므로 빅데이터에 적합
• 분산 데이터를 통해 24시간 365일 안전하게 데이터를 처리할 수 있는 고가용성 기능 제공

2.왜 카프카?

• 고가용성
  • 서비스에 지장없는 운영을 보장.
• 낮은 지연
• 확장성
• 높은 처리량
  • 높은 처리량을 감당하지 못한다면, 서비스를 유지하기 힘듦

RabbitMQ , Redis 와의 차이점

• 이벤트 브로커
  • 서비스에서 발생한 이벤트를 브로커의 큐에 저장함
  • 딱 한번 일어난 이벤트 데이터를 브로커에 저장함으로써 단일 진실 공급원으로 사용 및 재처리가 가능
  • 마이크로 서비스 아키텍쳐에서 중요한 역할을 함
  • kafka , kinesis
• 메세지 브로커
  • 대규모 메세지 기반 미들웨어 아키텍쳐에서 사용
  • RabbitMQ, Redis

카프카 구조

• 기존 1:1 매칭으로 개발하고 운영하던 데이터 파이프라인은 커플링으로 인해 한쪽 이슈가 생기면 다른 한쪽에도 영향이 간다. → 카프카는 이러한 의존도를 타파하였다. (디커플링)
• 큐에 데이터를 보내는 것이 프로듀서이고 큐에서 데이터를 가져가는 것이 컨슈머다

카프카 특징

• 높은 처리량
  • 높은 처리량을 감당하지 못한다면, 서비스를 유지하기 힘듦
  • 우리 비지니스의 성공여부는 어떤 Threash hold 에 걸쳐지면 안된다.
  • 파티션 단위를 통해 동일 목적의 데이터를 여러 파티션에 분배하고, 이런 파티션을 컨슈머로 병렬처리할수 있는것이 큰 특징
  • 파티션 개슈만큼 컨슈머 개슈를 늘릴수 있다
• 확장성
• 영속성
  • 파일 io 성능 향상을 위해 os 에서 담당하는 페이지 캐시를 이용한다. 그래서 파일을 쓰고 읽는데도 빠를수 있다.
• 고가용성

3.카프카 구성 요소

토픽

• 구체화된 이벤트 스트림 = 쉽게 큐로 이해하면 됨
• 하나의 토픽에 여러 Producer / Consumer 가 존재할 수 있다.
• 토픽은 담는 데이터에 따라 이름을 줄 수 있다.

컨슈머

• 기본적으로 가장 오래된 순서대로 가져감 - 0번 오프셋부터
• 새로운 컨슈머가 구독을 하게 되도 가장 오래된 순서대로 가져감
  • auto.offset.reset = earliest 인경우

파티션

• 카프카의 토픽들은 여러 파티션으로 나눠짐.
• 파티션의 끝에서 0번 부터 차곡차곡 쌓이게 됨
• 토픽 = 논리적인 개념이라면, 파티션은 물리적인 저장소에 저장하는 단위
• 각 파티션은 Append-only 방식으로 기록됨
• 특정 파티션으로 데이터를 쓸수 있고, 명시되있지 않으면 RoundRobin 방식으로 파티션을 배정한다
• 파티션을 늘린다면?
  • 파티션을 다시 줄일수는 없다.
  • 컨슈머 개수가 늘어날때 분산 처리할 수 있다.
  • 신규 데이터는 2개의 파티션 중어디로 들어갈까?
  • 보통은 라운드로빈으로 파티션을 할당함
  • 키의 해시값으로
• 파티션 삭제 주기는?
  • log.retention.ms : 최대 record 보존 시간
  • log.retension.byte : 최대 record 보존 크기

오프셋

• 각각 파티션의 레코드는 Offset 식별자 정보를 가짐, 데이터 번호
• 카프카는 메세지 순서를 보장 하지 않음. 하지만 파티션이 1개라면 보장할지도?

4.카프카 클러스터

카프카 클러스터

• 카프카 브로커
  • 설치된 카프카의 서버 단위
  • 보통은 3대로 구성
• replication
  • replication 이 1인 상태라면 파티션이 브로커 서버에 1개만 저장된다.
  • replicaion 이 2라면 원본 하나, 복제본 1개의 파티션이 각각의 브로커 서버에 저장된다.
  • 따라서 replication 개수 ≤ 브로커 서버 개수
  • 원본 파티션 = Leader 파티션, 복제본 파티션 = follow 파티션
  • replication 의 설정된 값에 따라 서로 다른 브로커 서버에 파티션의 복제본이 생긴다.
• ISR(In-SyncReplica)
  • 리터 파티션의 레코드 개수 만큼 팔로워 파티션의 개수가 동일하게 복제가 된 안정된 상태
• ACK
  • 카프카 프로듀서는 ack 를 이용해 고 가용성 보장
    • ack = 0 , response 무시. 속도는 빠르지만 유실이 있음.
    • ack = 1, reponse 를 받음, 파티션 복제는 보장 못함. 유실 가능성 있음
    • ack = all , response 받고, follwer partition 저장확인 절차를 거침. 유실 가능성 없음
  • replication 개수가 늘어난다면 성능 저하.
• 파티셔너
  • 데이터를 토픽에 어떤 파티션에 넣는지 결정하는 역할을 함
  • 메세지 키 또는 메세지 값에 따라 파티션이 결정됨
  • hash(키) = 파티션 넘버
• 카프카 lag
  • 운영시에는 consumer lag이 발생
  • lag 이란 = 컨슈머가 마지막으로 읽은 offset - 프로듀서가 마지막으로 넣은 offset
  • 한개의 토픽과 컨슈머 그룹에 대한 lag 이 여러개 존재 하게 된다.
  • max lag 에 대한 모니터링이 운영시에는 필요하다
• lag burrow
  • golang 으로 개발된 오픈 소스
  • 컨슈머 lag 모니터링을 도와주는 독립적인 애플리케이션
  • 멀티 카프카 클러스터 지원
    • 2개이상의 카프카 클러스터를 운영할때, 하나의 burrow 로 운영 가능
• 주키퍼
  • 코디네이션 애플리케이션
  • 브로커 서버와 통신하며 상태관리, 컨슈머와의 통신, 카프카 메타데이터 정보를 저장함.