매일매일 딥러닝

Kafka란?

Kafka는 분산 메세지 큐로 대용량 데이터를 다루면서도 빠른 데이터 처리가 가능하게 합니다.

본문의 내용은 데브원영님의 강의 '아파치 카프카 for beginners'를 수강하고 정리한 내용입니다.

ref: https://www.youtube.com/@DevWonYoung

Kafka Architecture

1. Kafa Producer API (source application) : click log, … logs 데이터 생성, 및 Topic에 데이터를 저장 (push)
2. Kafka Cluster(a.k.a Topic) : message 저장소 (핵심 역할)
3. Kafka Consumer API (target application) : Topic에 저장된 데이터를 읽음(pop). 로그 적재, 로그 처리
   • etc) Kafka Streams API, Kafka Connect API 

Apache Kafka Features

Topic & Partition

• Topic: 데이터가 저장되는 공간
  • 여러 개 생성 가능: 각 log별로 나눠서 저장 (message를 구분하는 용도)
• Partition: topic의 실제 저장 공간 단위
  • que 역할: Producer에서 들어온 데이터는 partition에 쌓이게 되고 Consumer에서 가장 오래된 데이터부터 가져감
    • 단, 데이터가 partiton에서 사라지지 않고 그대로 보관됨
    • (message가 저장되는 최대 시간과 크기를 지정할 수 있음)
  • partition 여러 개 생성 가능 (=병렬처리 가능)
    • 다중 partition의 경우 기본 저장 방법: 라운드 로빈
    • 단, partition은 늘릴 수 있지만 줄일 수는 없음
  • Offset: 메세지를 어디까지 넣었는지, 읽었는지 저장하는 값 (위치)
• Partitioner: Producer가 데이터를 보내면 partitioner를 통해 topic의 partition으로 데이터가 전송됨, partitioner는 데이터를 어떤 partition에 넣을지 결정하는 역할을 함
  • message key(hash)를 통해서 위치를 결정할 수 있음
  • 하나의 컨슈머는 하나의 파티션만 연결 (Consumer 1: Partition1), 하나의 컨슈머 기준으로 파티션의 순서 보장
• Lag: 모니터링 지표 중 하나
  
  • Consumer가 마지막으로 읽은 offset과 Producer가 마지막으로 넣은 offset의 Gap (여러 개 존재 가능)

Broker, Replication, ISR

• Broker: 카프카가 설치되어있는 서버 단위 (3개 이상을 권장)
• replication(핵심!): 서버의 이슈에 대비하여 데이터가 유실되는 것을 방지하고 복구하기 위해 복제본을 다른 broker에 저장해 둠 (=고가용성)
  • 최대 broker 개수만큼 지정할 수 있음
  • replication이 1인 경우, broker 중 1대에 partition의 정보가 저장됨
  • replication이 2인 경우, partition 정보가 원본 1개, 복제본 1개가 각각 2대의 broker에 저장됨

• ISR(In Sync Replica): **** 원본(leader)과 복제본(follower)을 합쳐서 부르는 용어
• Producer의 ack 동작: partition에 새로운 데이터를 저장할 때 리더 데이터의 업데이트와 팔로워 데이터의 복제가 정상적으로 되었는지 확인하는 방법 (option: 0, 1, all)

Zookeeper

: 분산처리 관리 시스템

• Kafka 관련 정보를 저장하는 역할
• Kafka의 Broker를 관리(유지보수, 조정)할 수 있음
• Apache Kafka에서 Zookeeper가 단계적으로 제거되는 중…

Consumer, Producer

여기서 자세히 다루지 않음. 더 궁금하신 분들은 아래의 두 동영상을 추천합니다!

• https://www.youtube.com/watch?v=geMtm17ofPY 
• https://www.youtube.com/watch?v=xqrIDHbGjOY 

장점

• High throughput message capacity
• Scalability & Fault tolerant : Broker, Partition, Consumer 등 구성요소를 쉽게 추가할 수 있음
• Undeleted log

설치, 실행하기

• Zookeeper (port: 2181, 2888, 3888)
• Kafka (port: 9092)
  • broker id: 각 서버 별로 다른 숫자로 설정하기
  • zookeeper의 connect 설정하기
• (java version)  https://www.youtube.com/watch?v=Qr0HVvtMFhg
•  (python version)  https://www.youtube.com/watch?v=VHngQ-moXIE

활용 예시

• https://techblog.gccompany.co.kr/apache-kafka를-사용하여-eda-적용하기-bf263c79efd0
• https://engineering.linkedin.com/blog/2019/apache-kafka-trillion-messages
• https://engineering.linecorp.com/ko/blog/how-to-use-kafka-in-line-1/

데이터 파이프라인 이란?

(a.k.a ETL(Extract-Transform-Load))

Data Source(log, api, product dataset,.. etc)에서 원하는 데이터를 Extract 해서 원하는 특정 format으로 Transform 하고 데이터를 원하는 곳(Data Warehouse,... etc)에 적제(Load)하는 일련의 과정 및 Architecture를 이야기한다.

(최근에는 순서를 바꿔 ELT 프로세스로 사용하기도 한다.)

멱등성이란?

데이터 파이프라인에서 가장 중요한 것은 재실행의 안전성이다. 즉, *멱등성(Idempotency)를 보장해야 한다. Airflow는 **Backfill이 쉽다.

*멱등성: 동일한 테스크를 여러 번 수행해도 동일한 결과가 되도록 하는 것 (도중에 오류가 발생해 재실행을 해도 중복이 발생하지 않아야 한다.)

**Backfill: 과거 데이터를 다시 채우는 과정

Airflow란?

워크 플로우 관리도구 중 하나로 데이터 파이프라인을 쉽게 해주는 프레임워크이다.

워크플로 관리 기술을 사용하면 스케줄러 기능뿐 아니라 데이터 파이프라인이 복잡해짐에 따라 크고 작은 장애가 발생하더라도 오류 처리와 다시 처리하기 위한 기능을 만들어 관리할 수 있다.

Airflow에 의한 워크플로는 여러 task(=operator)로 이루어진 DAG(Directed Acyclic Graph)의 형태로 정의한다.

예시

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

# config
dag = DAG(
    dag_id='helloword_dag',
    start_date=datetime(2022,12,19),
    catchup=False,
    tags=['example'],
    schedule_interval='0 2 * * *')  # crontab과 유사

def print_hello():
    print("hello!")
    return "hello!"

def print_goodbye():
    print("goodbye!")
    return "goodbye!"

# PythonOperator: 범용적
# 보통 task를 3개로 나눔 (ETL)
print_hello = PythonOperator(
    task_id='print_hello',
    python_callable=print_hello,
    dag=dag)

print_goodbye = PythonOperator(
    task_id='print_goodbye',
    python_callable=print_goodbye,
    dag=dag)

#Assign the order of the tasks in our DAG
print_hello >> print_goodbye

• 출처: '실리콘밸리에서 날아온 데이터 엔지니어링 스타터 키드' 스터디

Spark란?

: 인 메모리 형의 고속 데이터 처리

대량의 메모리를 활용하여 고속화 실현이 가능하다. Spark는 *Hadoop을 대체하는 것이 아니라 Hadoop의 구성 요소 중 분산 데이터 처리에 사용되는 **MapReduce를 대체하는 존재다. (Hadoop과는 다른 독립된 프로젝트이다.)

Spark는 스크립트 언어(JAVA, Scala, Python, R)를 사용할 수 있다는 장점이 있다.

*Hadoop: 대표적인 빅데이터 처리 시스템으로 분산된 다수의 컴퓨터에서 대량의 데이터를 처리하기 위한 시스템

**MapReduce: 대량의 데이터를 배치 처리하기 위한 시스템 (Map과 Reduce 기능으로 구성)

Pandas DataFrame vs Spark DataFrame

Pandas DataFrame

numpy ndarray를 기반으로 데이터 분석을 쉽고 빠르게 처리하기 위해 많이 사용되고 있다. 병렬 처리가 안된다.

이전 포스팅) 현업에서 많이 사용하는 Python 모듈 | Pandas 참고

Spark DataFrame

a.k.a SQL의 탈을 쓴 DataFrame

Spark는 SQL과 유사한 형태의 연산을 지원하는 API로 구성되어 있다.

병렬 CPU처리 및 분산 서버 데이터 처리가 가능하다.

Spark DataFrame의 경우 Pandas DataFrame과 달리 대부분 새로운 DataFrame 객체를 반환하는 형태로 구성되어있다.

예시)

table

Query

두 DataFrames간 변환이 가능하다.

spark_df = spark.createDataFrame(pandas_df)
pandas_df = spark.select('*').toPandas()

안녕하세요

오늘은 연관규칙 Apriori 알고리즘에 대해 알아보도록 하겠습니다!

Q1. Apriori 알고리즘 이란?

A1. 간단한 성능 측정치를 이용해 거대한 데이터베이스에서 데이터간의 연관성을 찾는 알고리즘

Q2. 그래서 어떤 패턴을 찾는데?

A2. 연관규칙은 장바구니 분석에 가장 많이 사용됩니다. 하지만 잠재적인 패턴을 찾을 때 또한 유용합니다.

예시: "기저귀를 살 때 맥주를 함께 산다"

그림출처: 사람과 디지털연구소

연관 규칙을 통해 맥주와 기저귀라는 별 상관없어 보이는 상품의 동시 구매 패턴을 발견하였습니다.

Q3. 굳이 왜 ML(머신러닝)으로 패턴을 찾아야해?

A3. k개의 아이템으로 2개씩 아이템을 묶어 연관패턴을 찾으려면 평가해야할 2^k개의 아이템 집합을 갖습니다.

만약 아이템이 100개면 2의 100승개의 아이템 집합이 생기므로 사람이 그 많은 데이터를 직접분석하기는 어렵습니다.

연관규칙에서 사용하는 두가지 통계척도

1. 지지도: 데이터에서 발생하는 빈도

count(x)
support(x) = ----------------------
N

2. 신뢰도: 예측능력이나 정확도의 측정치

                              support(x,y)
confidence(x -> y) = ------------------------------
                               support(x)

기저귀와 맥주의 예시 코드를 생성하여 연관규칙 알고리즘에 대해 알아봅시다!

# 1. 데이터 로드
x <- data.frame(
beer=c(0,1,1,1,0),
bread=c(1,1,0,1,1),
cola=c(0,0,1,0,1),
diapers=c(0,1,1,1,1),
eggs=c(0,1,0,0,0),
milk=c(1,0,1,1,1) )
# x 확인하기
x

# 2. arules 패키지를 설치한다. (apriori함수를 이용하기위해 설치)
# install.packages("arules")
library(arules)

trans <- as.matrix(x,"Transaction")
trans

# 3. apriori 함수를 이용해서 연관관계를 분석한다.
rules <- apriori(trans, parameter=list(supp=0.2, conf=0.6, target = "rules"))
rules

# support: 지지도, confidence:신뢰도, lift: 향상도, count
inspect(sort(rules))

결과

신뢰도가 클수록 연관관계가 높다는 의미입니다.

연관규칙을 평가하는 지수는 사실 위에서 언급한 2가지말고도 많은데

그 중 꽤 많이 쓰이는 것이 상관관계를 나타내는 lift입니다.

추가 : 연관관계 시각화하기

# install.packages("sna")
# install.packages("rgl")
library(sna)
library(rgl)

#visualization
b2 <- t(as.matrix(trans)) %*% as.matrix(trans) 
b2.w <- b2 - diag(diag(b2))
#rownames(b2.w) 
#colnames(b2.w) 
gplot(b2.w , displaylabel=T , vertex.cex=sqrt(diag(b2)) , 
vertex.col = "Pink" , edge.col="grey" , boxed.labels=F ,
arrowhead.cex = .3 , label.pos = 3 , edge.lwd = b2.w*2) 

결과

마지막으로, 연관규칙 학습을 위한 에이 프라이어리 알고리즘의 장단점을 알아보겠습니다.

장점1. 대규모 거래 데이터에서 작업할 수 있다.

장점2. 이해하기 쉬운 규칙을 생성한다.

장점3. '데이터 마이닝'과 데이터 베이스에서 예상치 못한 지식을 발굴하는 데 유용하다.

---------------------------------------------------------------------------------

단점1. 작은 데이터 셋에서는 유용하지 않을 수 있다.

단점2. 랜덤 패턴에서 비논리적 결론을 도출하기 쉽다.

오늘은 여기까지!!

다음은 k-means알고리즘으로 돌아오겠습니다!!