이것저것 👩‍💻

1. Airflow: "똑똑한 작업 반장" (Workflow Orchestrator)

Airflow는 수많은 데이터 작업이 언제, 어떤 순서로 실행될지 관리하는 도구

• 역할: "A 계열사 데이터를 가져온 뒤 -> B 계열사 데이터랑 합치고 -> 마지막으로 대시보드용 테이블을 만들어라" 같은 순서를 짠(DAG) 뒤, 정해진 시간에 실행

2. Trino: "엄청나게 빠른 요리사" (Distributed SQL Query Engine)

Trino는 여러 곳에 흩어진 데이터를 SQL 한 번으로 아주 빠르게 조회해 주는 엔진

• 역할: 데이터는 S3나 HDFS에 가만히 있고, Trino는 그 데이터를 메모리로 끌어올려서 순식간에 계산해. Redshift와 비슷하지만, 데이터를 직접 저장하지 않고 '계산'만 전담하는 게 차이점

3. 둘의 관계: "주문자와 요리사"

분석가들이 서비스를 사용할 때, 이 둘은 실과 바늘처럼 움직여.

1. Airflow(주문자): "자, 이제 분석할 시간이야! 이 쿼리 좀 실행해 줘."라고 Trino(요리사)에게 주문서를 던져.
2. Trino(요리사): S3에서 데이터를 슥슥 긁어모아 요리(계산)한 뒤, 결과를 다시 테이블에 저장하거나 Airflow에 알려줘.
3. K8s(식당 건물): 너는 이 둘이 싸우지 않고 자원을 잘 나눠 쓰도록 식당 건물을 관리(Config, Resource Limit)하는 주인장인 셈이지.

데이터 파이프라인을 잘 구축하는 것이 왜 중요할까?

데이터 파이프라인 구축이 중요한 이유는 데이터를 이동시키는데 더 적은 노력을 들일수록 핵심 비즈니스에 집중할 수 있기 때문이다.

카프카에서 데이터의 기본 단위는 메시지이며, 바이트의 배열로 이루어져있다. 메시지를 배치 단위로 저장하고 배치는 메시지들의 집합이다. 배치는 메시지를 쓸 때마다 네크워크 통신으로 발생하는 오버헤드를 줄이기 위한 것으로 지연량과 처리량 사이에 트레이드오프를 발생시킨다. 아파치 에이브로로 메시지 본체와 스키마를 분리한다. 스키마가 변경되더라도 코드 수정은 필요없다.

토픽과 파티션

메시지는 토픽 단위로 분류되며 토픽은 여러개의 파티션을 갖는다. 하나의파티션에서는 순서가 보장되지만 여러개의 파티션이 토픽 안의 전체 메시지에 대해서는 순서를 보장하지 않는다. 파티션은 카프카가 데이터 중복과 확장성을 제공하는 방법이기도 하다. 각 파티션이 서로 다른 서버에 저장될 수 있기에 하나의 토픽이 여러 개의 서버로 수평적으로 확장할 수 있고 파티션 복제도 가능하여 서로 다른 서버가 동일한 파티션의 복제본을 저장하고 있다. 분산 플랫폼인 이유이기도 하다.

프로듀서와 컨슈머

프로듀셔는 발행자, 작성자라고 불린다. 메시지 키와 키값의 해시를 특정 파티션으로 대응시켜주는 것은 파티셔너를 통해 구현된다. 커스튬 파티셔너도 가능하다. 컨슈머는 구독자, 독자라고 불린다. 컨슈머 그룹의 일원으로 작동한다. 이를 통해 대량의 메시지를 읽기 위해 컨슈머를 수평확장할 수 있다. 또한, 컨슈머 중 하나에 장애가 발생하더라도 그룹 안의 다른 컨슈머가 재할당받아 데이터를 읽어올 수 있다.

브로커와 클러스터

카프카 서버를 브로커라고 부르며 여러개의 브로커를 띄울 수 있다. 브로커는 클러스터의 일부이다. 하나의 클러스터 안에 여러 개의 브로커가 포함될 수 있으며 하나의 브로커가 클러스터의 역할을 하게 된다. 클러스터 역할이란 파티션을 브로커에 할당해주거나 장애가 발생한 브로커를 모니터링하는 등의 관리 기능을 담당한다. 브로커는 리더와 팔로워로 구분되며 복제 기능은 리더 브로커에 장애 발생 시 팔로워가 리더 역할을 이어받을 수 있게 하는 것이다. 프로듀서는 리더 브로커에 메시즈를 발행해야하고 컨슈머는 팔로워나 리더 중 하나로 데이터를 읽어올 수 있다.

다중 클러스터

미러메이커 툴을 사용해서 클러스터끼리의 데이터 복제도 가능하다. 기본적으로는 클러스터 내에서만 데이터 복제가 가능하다.

러닝 스파크 chapter 3장까지 읽은 내용을 정리해 보고자 한다.

HDFS와 맵리듀스 프레임워크를 사용하여 빅데이터를 처리하는 것에는 단점이 존재한다. 가장 큰 단점은 배치 작업을 수행하면서 MR 태스크의 중간 단계들을 로컬 디스크에 써야했단 점이다. 이것은 잦은 디스크 I/O 수행을 일으키며 데이터 처리 속도가 매우 느려진다.

그래서 아파치 스파크가 탄생하였다. 아파치 스파크는 대규모 분산 데이터 연산 처리를 위해서 설계된 통합형 엔진이다.

스파크는 하둡 맵리듀스의 단점을 보완하며 업그레이드 되었고 4가지 특징이 있다.

첫번째, 속도

스파크는 질의 연산을 방향성 비순환 그래프(DAG)로 구성하고 클러스터의 워크 노드 위에서 병렬 수행된다. 스파크 SQL 엔진 의 주요 기능 중 하나인 텅스텐(Tungsten)은 전체 쿼리를 하나의 함수로 합쳐 가상 함수 호출이나 중간 데이터를 위한 CPU 사용을 없앤다. 이 방법을 포괄 코드 생성이라고 한다. 이러한 방법은 CPU 사용량을 줄이고 효율을 높인다. 하둡 맵리듀스와 다르게 모든 중간 결과는 메모리에 저장되며 디스크 I/O 사용을 줄인다.

두번째, 사용편리성

데이터 프레임, 데이터세트, RDD API로 인해 사용자가 간결하고 단순한 코드 작성이 가능하다.

세번째, 모듈성

4가지 컴포넌트(스파크SQL, 스파크 정형화 스트리밍, MLlib, GraphX)가 존재하며 하나의 엔진 안에서 사용 가능하며 별도의 엔진을 돌릴 필요도 없고 별도의 API를 배울 필요도 없다.

네번째, 확장성

스파크는 빠른 병렬 연산에 초점이 맞춰져 있기에 저장을 담당하는 여러 데이터 소스와의 확장성이 뛰어나다. 하둡은 저장과 연산을 모두 포함하고 있다.

코드가 어떻게 스파크 애플리케이션에서 동작하는지 살펴보자.

애플리케이션

- API를 써서 스파크 위에서 돌아가는 사용자 프로그램. 드라이버와 클러스터 실행기로 이루어진다.

SparkSession

- 스파크 코어 기능들과 상호 작용할 수 있는 진입점을 제공하는 객체이며, 사용자가 SparkSession 객체를 생성해야 한다.

Job

- 스파크 액션에 대한 응답으로 생성되는 여러 태스크로 이루어진 병렬 연산

Stage

- Job은 스테이지라 불리는 서로 의존성을 가지는 다수의 태스크 모음으로 나뉜다.

Task- 스파크 이그제큐터로 보내지는 작업 실행의 가장 기본적인 단위

스파크 연산은 Transformation과 Action으로 나뉘며, Action이 호출되지 않으면 스파크는 아예 실행되지 않는다.

Transformation은 원본 데이터를 수정하지 않고 하나의 스파크 데이터 프레임을 새로 만들어서 결과값을 리턴한다.

모든 Transformation은 결과값이 즉시 계산되지 않는다. 이를 lineage라 하며, 기록된 리니지는 스파크가 트렌스포메이션들끼리 합치거나 재배열하거나 해서 효율적으로 실행할 수 있도록 최적화 한다. 트랜스포메이션을 리니지에 기록만 해두고 실행하지 않는 것을 지연 평가라고 한다. 코드에서 액션을 만났을 때 모든 트랜스포메이션의 지연 연산이 발동되며 새로운 데이터 프레임을 생성하여 작업한다.

=> 지연 평가는 쿼리 최적화를 가능하게 하고 리니지와 데이터 불변성은 장애에 대한 데이터 내구성을 제공한다.

좁은/넓은 트랜스포메이션

하나의 파티션 내에서 처리 할 수 있으면 좁은 트랜스포메이션으로 filter()나 contains()가 포함된다. 반면에 파티션끼리 데이터 교환이 발생하고 디스크 작업 등이 발생하면 넓은 트랜스포메이션으로 groupby()나 orderby()가 포함된다.

하둡 생태계 데이터 수집/저장/처리/적재
데이터 수집 - Kafka - 실시간 분산환경에서 메시지를 송수신하는 메시지 전달 솔루션
분산 데이터 저장 - HDFS - 데이터를 클러스터 환경에 분산 저장하는 솔루션으로 Namenode(리더노드)와 Datanode(컴퓨팅노드)로 관리
분산 클러스터 관리 - YARN - 분산 클러스터의 리소스 관리 솔루션으로 Resourse Manager가 Node Manager를 관리
분산 데이터 배치처리 - Hadoop MapReduce - Map과 Reduce의 2상로 데이터를 처리하는 하둡 기반의 배치 작업 플랫폼
인메모리 데이터 처리 - Apache Spark - 인메모리 상에서의 데이터 처리 플랫폼으로 배치처리, 실시간 스트리밍, SQL 질의와 Graph 처리, 머신러닝 같은 하위 프로젝트를 사용
데이터웨어하우스 연동 - Hive - 하둡 기반의 데이터 웨어하우스 시스템

분산 클러스터에서 데이터 검색 방법: ElasticSearch
Restful 프로토콜을 사용하여 정형, 비정형, 데이터 위치정보, 메트릭 등 원하는 방법으로 다양한 유형의 데이터 검색 수행

하둡 생태계를 사용하는 이유
1. 하둡의 HDFS는 동일한 저가 리눅스를 분산 클러스터로 사용하여 scale out을 사용 -> 저장소의 비용이 저렴함
2. 정형 데이터 외에도 반정형, 비정형 데이터 모두 다룸
3. YARN으로 메모리, CPU, 디스크 리소스를 효율적으로 저렴하게 관리 가능 -> 운영할때 리소스 관리 포인트에서 좀 자유로움
4. 딥러닝, 인공지능에서 필요한 INPUT 데이터 제공 가능 -> 데이터 모델링을 통한 데이터의 인사이트 추출 가능
5. Schema on Read -> 데이터를 read 하는 시점에 스키마를 지정하여 데이터 추출 -> 데이터가 저장되는 스키마의 위치가 모두 달라서 시간 절약 가능

데이터 전처리
정의: 원시 데이터를 비즈니스 요구사항의 분석 및 처리에 적합한 형식으로 데이터를 가공하고 처리하는 작업<br>
기법: 데이터 정제, 데이터 통합, 데이터 변환, 데이터 정리

ETL 프로세스(Extract/Transformation/Loading)
데이터 수집하고 비즈니스 규칙에 따라 데이터를 변환한 후 대상 데이터를 저장소로 적재하는데 사용되는 데이터 파이프라인

ELT 프로세스(Extract&Load/Transformation)
일단 데이터를 수집항 적재하고 이후 비즈니스 요구사항에 맞춰 데이터 변환 -> 비즈니스 요구사항이 정해지지 않았어도 데이터 적재 가능

하둡파일시스템
- 구성: 마스터 역할의 NameNode와 데이터를 실제로 저장하는 다수의 슬레이브 Datanode
- 구조: 마스터-슬레이브 구조(하나의 마스터에 여러개의 슬레이브 구조)
- 동작원리
  1. 사용자가 데이터를 HDFS에 저장
  2. HDFS Client는 메타데이터(파일명, 폴더구조 등)를 NameNode에 저장
  3. NameNode는 Datanode에게 데이터 관련 명령 지시,  Datanode는 자신의 상태 정보를 NameNode에게 송신
  4. Datanode안의 여러개의 블록으로 나눠져 실데이터가 복제 저장됨
  5. 사용자가 데이터를 읽어올 경우, HDFS Client는 NameNode로부터 파일 블록에 대한 메타 데이터 정보를 수신
  6. HDFS Client가 원하는 데이터가 있는 Datanode의 파일 블록을 찾아서 파일을 읽어옴.
  7. NameNode의 메타 데이터 변경 시, Secondary Namenode의 Edits 파일에 로그 저장
 
Yarn의 구조
|데몬|기능|
|------|---|
|Resource Manager|Yarn의 마스터 역할 수행(총괄 관리자)|
|Scheduler|Resource Manager 내의 데몬으로 Node Manager와 통신하며, Application Master에 자원 할당|
|Application Manager|Resource Manager 내의 데몬으로 Application Master 실행 및 상태 관리|
|Node Manager|각 노드의 상태 정보를 모니터링 하여 Resource Manager에게 제공|
|Application Master|각각의 Application을 관리하는 마스터의 역할|
|Container|Application Master가 관리하는 태스크를 수행, Node Manager에 의해 생성|

하둡파일시스템의 장점
- 스케일 아웃 분산 시스템(병렬로 스케일 아웃됨으로 데이터 양에 따라서 서버 필요)
- 장애복구(마스터 서버가 슬레이브 서버의 장애를 실시간으로 감지하여 서비스 우회 OR 다른 노드에 복제된 데이터 사용)
- 대용량데이터(한번 저장된 데이터는 only read = 데이터 업데이트 X)

FRED 데이터를 가져오기 위해서 파이썬 라이브러리인 fredapi를 설치해야한다.

그런데 아나콘다 가상환경 모듈에서 위와 같은 오류가 발생해서 확인해보니 Python 3.9 이상에서는 fredapi에 저런 함수는 사용하지 않는다고 한다. 즉 파이썬 버전과 fredapi 라이브러리의 충돌이다. 그래서 fredapi를 재설치를 해줘야했다.

그래서 먼저 아나콘다에 설치되어 있는 패키지를 삭제해줬다.

conda uninstall fredapi

문제는 conda install fredapi를 하면 PackagesNotFoundError를 발생시켜서 설치가 안된다.

그래서 pip을 사용해서 설치해줬더니,,, 이번엔 pandas에서 오류가 나서 pandas도 재설치해줬다.

pip install fredapi

다시 한번 느끼지만 파이썬은 버전에 너무 민감해서 버전 관리를 하는 게 항상 문제인 거 같다.