Spotify API를 활용해서 데이터를 꾸준히 모으고, 이를 시각화해서 볼 수 있는 대시보드를 구축하는 프로젝트를 했다. MLSA가 된 덕분에 매달 Azure $150 크레딧이 생겨서 평소에는 꿈도 못 꿨던 비싼 거 써봤다.

이 중에서 제일 비싼 서비스는 Databricks. 특히 웨어하우스를 쓰려면 Premium으로 사용해야 했는데, 14일 DBU 무료 체험을 할 수 있다. 가격이 얼마나 나올까 궁금해서 하루만 Standard로 써봤는데, Spark job 돌리면 DBU 가격이 2만원 나왔다. ㅎㅎ 심지어 Premium도 DBU만 무료고 VM 가격은 따로 받는다. 사용하지 않을 때 종료시켜놓고, 자동 종료시간 맞춰놔야 불필요한 돈이 나가지 않는다.

Azure Data Factory

전반적으로 Airflow로 스케줄링을 했는데, 사실 간단하게 하려면 Azure Data Factory만 써도 됐을 것 같다.

Pipelines

아키텍처처럼 데이터 팩토리 안에 파이프라인 2개를 만들었다. 이 파이프라인의 역할은, 내가 의도한 건 사실 AWS Glue의 catalog 같은 느낌이었다. 그래서 파이프라인 이름도 catalog로 지었다.

AWS와 Azure 서비스 비교에 관해 Azure에서 공식적으로 정리해둔 문서가 있다. Azure Purview 아니면 Data Factory라길래 Purview부터 사용해봤는데, 내 스타일이 아니었다.. 그래서 Data Factory로 갈아탔는데 가격도 저렴하고 마음에 든다.

파이프라인 하나만 예시를 들자면 이렇게 구성했다.

PostgreSQL 서버에 쿼리를 날리기 위해 Web Activity를 추가했다. 쿼리 날리는 작업은 따로 구성해둔 백엔드 서버에서 수행한다. 그냥 이 Activity는 요청을 날려서 성공 응답을 받으면 다음으로 넘어간다.

나머지는 모두 Copy Data Activity다. Blob에 있는 *.json 파일을 읽어들여 Sink - Mapping 사전 세팅에 따라 PostgreSQL에 복사한다.

특히 *.json이나 *.parguet의 Copy Data Activity의 경우 이렇게 구성하면 AWS Glue의 카탈로그와 비슷한 느낌을 낼 수 있었다. AWS는 S3에서 날짜 파티션을 year=2024/...의 바로 바깥 폴더로 지정해 주면 알아서 파티션을 인식한다.

Azure Data Factory에서도 마찬가지다. 다만 두 번 작업을 해줘야 한다. 먼저 이렇게 year=2024 바깥 폴더를 지정해서 데이터셋을 만들어준다.

데이터셋을 불러오고 추가로 와일드카드 경로 입력을 통해 불러올 파일을 확정 지어야 한다. File path type 옵션에서 기본 값인 File path in dataset하면 와일드카드 지정하라고 오류가 뜨기 때문이다.

Spark

데이터 전처리를 위해 Spark를 쓰고 싶었는데, AWS의 경우 Glue 안에서 노트북, 파이썬으로 Spark 작업이 가능하지만 Data Factory는 다른 서비스에서 작성한 노트북, 파이썬을 Activity로 연결해야 했다.

Spark를 위한 Activity 선택지는 많았다.

1. Databricks
2. Synapse (이건 Data Lake Storage를 안 써봐서 아직도 잘 모르겠음)
3. HDInsight

이 과정에서 제일 시간을 많이 잡아먹었다. Spark 코드 자체는 금방 짰는데, 어떤 서비스를 써야 가장 좋을까 고민하면서 많이 시도해봤기 때문에 🙂

처음에는 HDInsight 쓰려고 했다. 그래서 quota 요청도 해서 부족한 만큼 받아내고, 설레는 마음으로 서비스를 시작했는데 세상에... 너무 느렸다. 시작될 때까지 계속 기다리는데, 이러다가는 매일 수행되는 작업이 이렇게 오래 걸려서 되겠나 싶었다.

Databricks

그래서 Databricks로 갈아탔다. ㅋㅋㅋ 몰랐는데 Azure에만 있는 게 아니라 AWS, GCP에도 서비스 되고 있더라. 공식문서가 Authentication 관련 내용 제외하고는 똑같은 내용으로 Microsoft, AWS 두 Document에 존재한다.

Databricks를 사용하게 된 계기는 다음과 같다.

1. Data Factory에 Spark Activity를 넣어야하는데 제일 만만했다.
2. 14일 무료체험해서 개인 프로젝트 수준인 나에게는 가격 부담이 덜했다.
3. 프리미엄이라면 Spark뿐만 아니라 Warehouse도 사용할 수 있었다.

그리고 이 댓글이 가장 나를 Databricks로 입문하게 만들었다.

Spark

Compute에 Cluster를 사용하면 PySpark 코드를 돌릴 수 있었다.

SQL Warehouse

Amazon Redshift Spectrum의 외부 테이블 느낌을 내고 싶었는데, 잘한 건가 모르겠다. 좀 산으로 간 느낌?

아키텍처 그림에서처럼 PostgreSQL에 있는 데이터를 SQL Warehouse로 주기적으로 옮기기 위해 정말 많은 사전 세팅을 했다.

Fivetran을 이용해 데이터를 주기적으로 sync 했다. 회원가입 따로 해야 한다.

처음 연결하기 위해서 PostgreSQL 서버 정보를 입력해야 한다. 그리고 제일 중요한 작업. Setup Guide 못 보고 저장하려 하면 계속 실패한다. 왜 실패할까 고민하며 정말 많은 시간을 들였는데, 등잔 밑이 어둡다고... 각자 선택하고 싶은 Update Method에 맞는 Setup Guide를 읽고 따라 한 뒤에 저장해야 했었다. 앞으로는 글을 잘 읽자.

sync 시간을 지정할 수 있었다. 24hours를 고른 다음에 시간을 지정할 수 있다. 시간 지정하려고 API까지 써서 했는데, 홈페이지에서 할 수 있었다. 한참 찾았다. UTC 아니고 PST니까 조심하자.

Airflow Connections

나의 경우 지금 기준 가장 최신 버전인 Airflow 2.8.1을 사용했는데, Data Factory는 Connection Type 목록에 기본으로 있었지만, Databricks는 없었다. 그래서 docker-compose.yaml파일에 다음과 같이 추가했다. _PIP_ADDITIONAL_REQUIREMENTS: ${_PIP_ADDITIONAL_REQUIREMENTS:- apache-airflow-providers-databricks}

# 일부만 가져옴
from airflow.providers.databricks.operators.databricks_sql import DatabricksSqlOperator
from airflow.models import Variable

databricks_sql_endpoint = Variable.get("databricks_sql_endpoint")

create_schema_task = DatabricksSqlOperator(
    databricks_conn_id='databricks_conn',
    sql_endpoint_name=databricks_sql_endpoint,
    task_id='run_dw_create_schema',
    sql='create_schema.sql',
    dag=dag,
)

Databricks SQL 웨어하우스 설정에서 아무리 찾아봐도 엔드포인트는 없었다. 알고 보니 그냥 내가 지정한 SQL 웨어하우스 이름 그 자체가 엔드포인트였다..🤦‍♀️

그리고 만약 토큰 관련 오류가 난다면 이 글을 참고하면 좋을 것 같다. token을 이미 한 번 Connections에서 지정했어도, Extra field에서 한 번 더 지정해 줘야한단다. (왜일까...?)

고찰

이 프로젝트를 기점으로 Azure 서비스를 많이 써봐서, 내게 여러모로 도움이 되었다. AWS랑 비교하기도 해보고, Azure 서비스에 관한 자료가 많이 없어서 계속 찾아보고 이것저것 다 시도하고😏 고생은 많이 했지만 다 해결해서 뿌듯하다.

• Azure Container App
• Docker Hub
• GitHub

개요

Microsoft Student Ambassador가 되고 나서 Alpha 마일스톤으로 승급했다. 덕분에 Azure $150를 매달 지원받는데, 안 쓰면 아깝다. 기존에 University Student로서 지원받았던 계정에는 대략 5만원이 남았다. 빨리 해치우고 엠버서더 계정으로 갈아타야지! 하는 생각에 이 CI/CD 여정이 시작되었다.

구조

1. GitHub Repository의 특정 폴더 안에서, 혹은 특정 브랜치에서 커밋 push가 일어날 때마다 Dockerfile을 빌드하고 이미지를 Docker Hub에 올린다.
2. Docker Hub에 이미지가 update될 때마다 Azure Container App에서 pull 하고, run한다. (주기적으로 Container App에서 이미지 변경을 확인하려면, Container App Job을 만들어서 Cron 식을 입력해 일정 시간동안 트리거해야하는 것 같다.)

상세 설명

먼저 GitHub repository 안에 아래처럼 .github/workflows/ci.yml이라는 파일을 만들었다고 가정하자.

name: Docker Image CI

on:
  push:
    branches: [ "main" ]
  pull_request:
    branches: [ "main" ]

jobs:

  build:

    runs-on: ubuntu-latest

    steps:
    - uses: actions/checkout@v3
    - name: docker login
      env:
        DOCKER_USER: ${{secrets.DOCKER_USER}}
        DOCKER_PASSWORD: ${{secrets.DOCKER_PASSWORD}}
      run: |
        docker login -u $DOCKER_USER -p $DOCKER_PASSWORD
    - name: Build the Docker image
      run: docker build --tag ${{secrets.DOCKER_USER}}/scraper:latest -f my_server/Dockerfile .
    - name: docker push
      run: docker push ${{secrets.DOCKER_USER}}/scraper:latest

여기서 이미지 이름은 scraper이고, 태그 이름은 latest이다. 그리고 Dockerfile이 있는 폴더명은 GitHub repository 루트로부터 my_server/Dockerfile이다. → Dockerfile이 루트 디렉토리에 있는 게 아니라면 Dockerfile 내에서도 COPY할 때 신경 써야 한다. COPY . .라고 하던 거를 COPY /my_server/. . 해야 함.

Secret 값들은 GitHub Settings에서 미리 넣어두어야 한다.

위처럼 workflow를 만들어주면 변경사항이 있을 때마다 Docker Hub에 올려주는 CI가 된다.

그리고 이제 Azure Container App을 하나 만든다.

연결 대상을 이처럼 Docker Hub로 설정하고, 이미지 이름과 태그에 {유저명}/{이미지명}:{태그명}으로 입력하여 연결하면 된다.

여기서 잠깐. 나는 Dockerfile에서 EXPOSE 8000을 하였다. 그런데 Azure Container App에서는 어떻게 접속하면 될까?

이렇게 수신 탭에 들어가서 HTTP 포트를 뚫어준다. 사진에는 잘렸지만 수신 트래픽도 '어디서나 트래픽 허용'으로 선택해야 한다. 그러면 엔드포인트로 들어갈 시 정상적으로 배포된 모습을 확인할 수 있다. 이렇게 CD도 완료.

Q. What role does a pod play in a vector database index? A. It represents pre-configured hardware for efficient data retrieval

Collection

• Static copy of a index
• collection에서 query 할 수 없음
• index 백업에 유용함

Create collection

pinecone.create_collection(name = "my-collection", source = "test")

List, Describe, Delete collection

• List

  pinecone.list_collections() # list all collections in database

• Describe

  res = pinecone.describe_collection("my-collection")
  res.name # 결과는 'my-collection'
  res.size / 10**6 # MB단위, 결과는 3.112836

• Delete

  pinecone.delete_collection("my-collection")

Namespace

• Index는 한 개 이상의 namespace를 가질 수 있음
• 모든 vector는 무조건 한 개의 namespace에 있어야 함 (Pinecone의 namespace default 값은 "")
• namespace 값은 index 안에서 고유해야 함

'''
upsert할 때 namespace를 지정하는 예시
'''
idx = pinecone.Index("my-collection-index") # connect to index

### 기타 코드 생략 ###

idx.upsert(vectors_subj, namespace='subject')

'''
query할 때 namespace를 사용하는 예시
'''

### 기타 코드 생략 ###

idx.query(vector = list(np.random.rand(3)), 
          top_k=3, 
          namespace='',
          include_values=True)

Metadata

• vector와 관련된 추가 정보 개념. 필터링, 정렬할 때 활용할 수 있음

Upsert

여기서 {"topic": ~~~~, "year": ~~~~} 부분이 metadata.

idx.upsert([
    ("1", [0.1, 0.1, 0.1 ], {"topic": "subject", "year": 2023}),
    ("2", [0.2, 0.2, 0.2], {"topic": "other", "year": 2024}),
    ("3", [0.3, 0.3, 0.3], {"topic": "body", "year": 2023}),
    ("4", [0.4, 0.4, 0.4], {"topic": "body"}),
    ("5", [0.5, 0.5, 0.5], {"topic": "subject"})
])

Query

idx.query(vector =[0,0,0], 
          top_k=2, 
          include_metadata=True, 
          include_values=True,
          filter={
             "topic" : {"$eq": "subject"},
              "year" : 2023
         })

• DB와 Vector DB의 관점에서 CRUD를 정리한다.

Pinecone

• Jupyter Notebook에서 실행할 때의 코드.
• 경고 문구 제거하려면 from tqdm.autonotebook import tqdm

  Create

  import pinecone
  pinecone.init(api_key="xxx", environment="xxx")
  pinecone.create_index(name="insert", dimension=3) # insert라는 이름의 db 생성
  pinecone.list_indexes() # 윗줄이 정상 실행되면 ['insert']가 출력될 것임.
  

vectors = [[1, 3, 4], [5, 6, 7], [8, 9, 0]] vect_ids = ['vec1', 'vec2', 'vec3'] idx = pinecone.Index('insert') idx.upsert([ ('vec1', [1, 3, 4]), ('vec2', [5, 6, 7]), ('vec3', [8, 9, 0]) ]) # list of tuples. 결과는 {'upserted_count': 3}

### Update
```python
# 코드 생략
idx = pinecone.Index('insert')
idx.upsert(vectors = [
    ("A", [0.1, 0.1, 0.1]),
    ("B", [0.2, 0.2, 0.2]),
    ("C", [0.3, 0.3, 0.3]),
    ("D", [0.4, 0.4, 0.4]),
    ("E", [0.5, 0.5, 0.5])
]) # 결과는 {'upserted_count': 5}

idx.update(id="E", values=[0.55, 0.55, 0.55]) # 특정 id에 바꾸고 싶은 val로 설정

Query

# 코드 생략
idx.upsert([
    ('vec1', [1, 3, 44]), # update because vec1 is already in DB
    ('vec2', [5, 6, 77]), # update
    ('vec33', [8, 9, 0]) # insert
])

idx.query([0, 0, 0], top_k=5, include_values=True)

query 결과는 {'matches': [{'id': 'vec3', 'score': 0.0, 'values': [8.0, 9.0, 0.0]}, ...], 'namespace': ''}

Fetch

# 코드 생략
idx.fetch(ids=["vec1"]) # ids=["vec1", "vec2"] 이렇게도 가능

fetch 결과는 {'namespace': '', 'vectors': {'vec1': {'id': 'vec1', 'sparse_values': {'indices': [100, 200, 300], 'values': [0.5, 0.5, 0.5]}, 'values': [0.1, 0.2, 0.3]}}}

Delete

# 코드 생략
idx.delete(ids=['vec33'])

단, 없는 id로 삭제하려고 해도 에러는 발생하지 않음

idx.delete(delete_all=True) # 모두 삭제

• Index: DB와 비슷한 개념
• 프리티어는 1개 Index, 1개 Project만 생성 및 사용 가능
• API Key 발급하여 사용

Ubuntu에서 환경설정

• python venv 디렉토리 안의 pyvenv.cfg 내용 중에

  home = /home/{내_컴퓨터_이름}/.pyenv/versions/3.8.2/bin
  include-system-site-packages = false
  version = 3.8.2

여기서 false를 true로 변경

• pinecone 설치

  pip3 install pinecone-client

• pinecone 사용 기초

  

import pinecone

pinecone.init(api_key="내_API_KEY_Value", environment="gcp-starter")
pinecone.create_index(name="hello-pinecone", dimension=4)

• (참고) Jupyter Notebook에서 Shift+Tab키를 누르면 현재 함수 또는 객체에 대한 도움말 팝업 표시됨. 파라미터 설명 확인 가능

Udemy의 Master Vector Database with Python for AI & LLM Use Cases 강의를 공부하고 정리하는 글.

학교에서 자연언어처리 시간에 교수님께서 잠깐 소개하고 넘어가신 Vector DB. 개인적으로 흥미로워서 유데미 강의 목록에 추가해두었다. 방학이 되어서야 듣기 시작한다. 근데 강의가 자막이 하나만 있는데 영어 자동이다... 🙂

Vector DB

A specialized type of DB designed to efficiently store manipulate high-dimensional vector data.

Vector Embedding

ex) 2차원의 이미지 → 임베딩 모델 → [0.12, 0.23, 0.5, ...] 차원이 증가하면 벡터 거리 계산법이 더 복잡해짐

Vector DB Use Cases

• 최근 유행
• Fixing Hallucination Problem of LLMs with source knowledge
  • Hallucination Problem은 Large Language Model이 트레이닝 데이터에 의존하기 때문에 발생하는 문제.
• Semantic search & Similarity search: 오디오, 비디오, 이미지, 텍스트
• 추천 시스템
• 머신러닝, 이상치 탐지, 그래프 분석 등

Example of Vector DB

• Pinecone
• Redis
• PgVector
• ScaNN

오랜만에 글을 써본다. 하루 종일 삽질하고 쓰는 후기

Azure Student 구독을 사용하면 Azure Speech Service를 무료로 사용해볼 수 있다.

프로젝트에서 목표가 STT -> GPT -> TTS 였는데, API 최초 탐색 후 선정은 아래와 같았다.

STT: ETRI 음성인식 OPEN API TTS: 네이버 클라우드 CLOVA Voice - Premium

그런데 ㅋㅋ ETRI 음성인식 API를 써봤더니 "나 오늘 학교에서 그림 잘 그렸다고 칭찬 받았다?"를 말했더니 외계어를 쏟아낸다... 성능이 아쉽다. 🤨

그리고 아무 생각 없이 네이버 클라우드 가입하고 십만 원 크레딧 받았더니, 정작 내가 쓰고 싶은 TTS 서비스는 크레딧 적용이 안되는 항목이었다.. ㅋㅋ? 그래도 언젠가는 쓸 날이 올 거야~

내가 Azure를 쓰는 이유

그래서 Azure 서비스를 찾다가 Speech Service는 써보니까 성능도 좋고, 반응 속도도 빠르고, 무엇보다 무료라... STT뿐만 아니라 TTS도 할 수 있어서 그냥 한꺼번에 하면 된다.

나는 Azure의 Document만 보면 겁이 나던데, 뭐랄까 번역이 덜 된 부분도 가끔가다 있고, 여기저기 이동하면서 다시 찾아야 하는 내용도 많고,, 무엇보다 인지도가 그렇게 높지 않아서 한글 자료도 별로 없다.

그럼에도 불구하고 내가 Azure를 좋아하는 이유는 유용한 무료 서비스가 정말 많다는 것. 특정 사용량까지는 무료고, 그 이상으로 넘어가면 그때부터 무료 크레딧에서 까이는 서비스들이 많아서 고맙다. (단 하나 아쉬운 점은 Azure Students 구독으로 GPU는 쓸 수 없다는 것)

그리고 특히 이 Speech Service는 굉장히 편리하다. 나는 당연히 텍스트를 주면 음성 파일로 저장하거나, 음성 파일이 존재하는 걸 불러와서 텍스트로 변환하는 건가 했는데. 그게 아니라 파일 없이 실시간으로 해준다.

1. STT: 최대 30초까지 말할 수 있고, 알아서 침묵이 느껴지면 듣기를 종료한다. 😮
2. TTS: 그냥 자기 할 일 끝나면 바로 스피커 쪽으로 소리 출력해버린다. 😎

어떻게 하는가

무료 F0 요금제로 인스턴스를 만들고, speech key, service region, 성우 이름만 알면 된다.

여기에 보이는 키1이 speech key이고, 위치/지역이 service region이다. 나는 한국말하는 성우 중에 어린이 목소리에 가까운 서현을 골랐다.

Azure에 언어별 샘플 코드가 웬만하면 다 있으므로 확인해보시고,, ! Python 코드를 Class로 짰는데 그중에 일부만 가져왔다.

import azure.cognitiveservices.speech as speechsdk

# ...

def tts(self):
    """
    Azure Speech Service를 이용해 Text-to-Speech
    """
    speech_key = self.azure_speech_key
    service_region = self.azure_service_region

    speech_config = speechsdk.SpeechConfig(subscription=speech_key, region=service_region)

    speech_config.speech_synthesis_voice_name = "ko-KR-SeoHyeonNeural"

    if self.answer_text == "":
        text = "오류가 발생했어요. 문제가 계속된다면 관리자에게 문의하세요."
        logging.error("TTS API 실행 전 Answer Text Result가 비어있습니다.")

    text = self.answer_text

    speech_synthesizer = speechsdk.SpeechSynthesizer(speech_config=speech_config)

    result = speech_synthesizer.speak_text_async(text).get()

    if result.reason == speechsdk.ResultReason.SynthesizingAudioCompleted:
        logging.info("TTS Speech synthesized for text [{}]".format(text))
    elif result.reason == speechsdk.ResultReason.Canceled:
        cancellation_details = result.cancellation_details
        logging.error("Speech synthesis canceled: {}".format(cancellation_details.reason))
        if cancellation_details.reason == speechsdk.CancellationReason.Error:
            logging.error("TTS Error details: {}".format(cancellation_details.error_details))

alsa

ㅎㅎ 이렇게 생각보다 API 적용은 금방 했지만, 정작 라즈베리파이에 연결된 스피커와 마이크가 말썽이었다. python3 test.py 이런 식으로 실행하면 아무런 문제 없이 되던 프로그램이, systemd에 올린 뒤 실행하니 마이크랑 스피커를 인식하지 못한다... 🙃 너 갑자기 왜 그래

또 열심히 구글링 해보니 나랑 똑같은 사례가 존재했다.

cat /proc/asound/cards를 통해 스피커와 마이크가 연결된 id를 알아내고, 그 id에 맞게 /etc/asound.conf에 아래와 같이 입력해 주면 해결된다. 나의 경우 pcm은 마이크였다. 아직도 ctl이 스피커인지는 모르겠다. 잘 돼서 따로 확인을 안 해봤다.

 defaults.pcm.card 3
 defaults.ctl.card 0

.wav 효과음 스피커로 출력하기

이거도 예상외로 굉장히 시간을 많이 잡아먹었는데.. 단순히 .mp3 또는 .wav 파일을 pygame으로 소리 출력하면 되지 않을까? 싶어서 했는데 ALSA 관련 오류가 나면서 계속 실패했다. 솔루션을 찾아봐도 안 나와서 포기할까 했지만, 이건 비정석적인 방법을 동원해서라도 해결해야겠다 싶었다.

os.system을 사용해서 코드가 안 예쁘긴 하지만.. 이렇게라도 해야만 했다. cat /proc/asound/cards를 해서 나오는 오디오 잭 id를 0으로 알아냈고, 그 결과 아래처럼 코드를 작성해 해결했다.

import os
os.system(f"/usr/bin/aplay -D hw:0,0 {파일명.wav}")

끄읕

따끈따끈 끝난지 얼마 지나지 않은 프로그래머스 데브코스 데이터엔지니어링 후기를 적어보려고 한다. 벌써 이게 추억이 되어버렸다 🤣

지원금

매달 기본 훈련수당 + KDT 훈련수당으로 인해 지원금은 30만원 정도 받았다. 나는 4학년 막학기가 아니라 국취제 대상은 아니었다 ㅎ.. 그래도 후회는 없다.

수강생

• 연령 1기에서는 수강생분들의 나이대가 굉장히 다양하다고 들었다. 나이 때문에 걱정은 안하셔도 될 듯.

• 열정 대분류로 따지면 팀이 6개로 나뉘었는데, 난 다른 팀과의 교류가 없어서 타팀은 모르겠지만. 우리 팀은 진짜 열정이 엄청났다...

  데브코스 방학, 주말, 공휴일 안 가리고 프로젝트 했다. 정말 팀원들에게 많이 배웠다. 대분류 팀은 랜덤 배정이었는데 운이 좋았다.

• 인원 총 60명이었는데 맨 마지막에 봤더니 절반만 남았다. 취업 또는 개인 사정 때문일텐데, 사실 처음에 60명은 너무 많아보였는데 이런 경우를 고려해서 많이 뽑지 않았나 싶기도 하고.

강사진과 멘토님

• 강사진 데이터엔지니어링 분야 강사님께서 진행하는 강의 비중이 제일 큰데, 이분은 단언컨대 최고였다. 매료되는 강의 감사했습니다.. 😆 Django 파트 맡으신 분의 강의도 새롭게 배운 점이 많아서 좋았다. 근데 다른 분들은 오래 돼서 기억이 안 난다..

• 멘토님 팀별로 멘토님이 1명씩 배정되는데, 내가 속한 팀은 멘토님이 중간에 한 번 바뀌셨다. 두 분 모두 정말 좋은 분이셨고 엄청난 전문가셨다. 멘토님께서 팀별 프로젝트에 대해서 첨언을 해주시는 시간이 있었다. 이 시간이 정말 많은 도움이 되었고, 이때가 데브코스 하길 정말 잘했다고 생각했던 순간 중 하나다.

  그 외에도 개별적으로 진로 멘토링 시간도 매주 있었는데, 대학 졸업까지 시간이 복학하고도 1년 반 남은 나에게는 매번 새로운 질문을 생각하는 게 꽤나 어려웠다.. ㅋㅋㅋ 이 프로그램 자체가 취업을 위해 만들어진 거니까, 취준생이라면 정말 많이 도움 되실 거다.

  Slack 질문 게시판에 질문을 올리면 타팀 멘토님께서 답변해주시기도 한다. 타팀 멘토님의 특강을 들을 수 있는 시간도 있었다. 하지만 거의 타팀 멘토님과는 교류할 일이 없었다.

프로젝트

데이터엔지니어링에 관한 프로젝트를 총 4번 했다. 4번 다 사용해야 하는 기술 스택 틀이 정해져 있지만, 주제는 자유주제다. 3개는 일주일 안에 완성해야 하는 미니 프로젝트였고, 마지막 1개는 한 달동안 진행되는 최종 프로젝트였다.

5개월이라는 짧다면 짧고 길다면 긴 시간 동안 프로젝트가 4번이나 있어서 겁을 먹었지만, 적응되면 할 만하다. 앞선 미니 프로젝트는 최종 프로젝트를 위한 연습이라고 생각하면 편하다.

AWS 지원은 최종 프로젝트 때만 지원받았는데, 다음 기수는 어떻게 되려나 모르겠다. 확실히 AWS를 지원받고 프로젝트를 하니 퀄리티가 높아진 게 확 체감이 됐달까? 데브코스 덕분에 AWS의 다양한 서비스를 경험해보고 프로젝트에 적용할 수 있어서, 이때가 데브코스 하길 잘했다고 생각했던 또 다른 순간.. ㅎㅎ

강의에서 배운 걸 실제 프로젝트로 적용하는 걸 겪어보는 일도 소중했지만, 프로젝트를 함께 하면서 협업하는 법, 클린 코드, PEP8, GPT 활용, 알아볼 수 있게 & 효율적으로 코드 짜기 등등의 중요성을 깨달을 수 있었다.

대면? 비대면?

1기의 경우는 최초에 공지할 때 최종 프로젝트는 대면으로 진행한다고 되어 있었다. 그래서 지방에 거주하는 나는 고시원을 알아봐야 하나, 친구 집에 얹혀살아야하나 생각했는데, 일단 지원하고 봤다.

붙고 들어갔더니 대면 → 비대면 → 비대면 + 대면 혼합으로 방식이 바뀌었고 나는 고민에 빠졌다. 개인적으로 학교에서 할 게 아직 남아서 결국 비대면을 택했다.

이건 좀 아쉬움이 남는 게, 대면을 선택했다면 열정 넘치는 우리 팀원들, 타팀 수료생과 친해질 수 있는 시간이 되지 않았을까.. 🥺 우리 팀,, 맨 마지막 발표회 때만 뵈어서 넘넘 아쉽다.

수료 후기

휴학 중에 시간이 맞는 부트캠프를 찾다가 합류하게 된 데브코스! 원래 데브코스가 여러분도 아시다시피 유명해서 내가 들어갈 수 있을까? 싶었지만 운 좋게 들어가서 말로 형용할 수 없을 만큼 사람들에게 많이 배웠다. 안 했으면 어쩔 뻔..!!

개강하고 난 뒤에 수료 행사해서 아직 수료증 못 받았는데 언젠가 받으면 여기에 올려둬야지 ^^.. 기껏 열심히 배운 거 안 까먹게, 계속 데이터엔지니어링 공부를 이어나가야겠다.

준비물

준비물은 다음과 같다. 대학생이면 1년 무료다.

1. GitHub Student Developer Pack의 Namecheap 또는 Name.com
2. Azure for Students의 VM Linux.

근데 Namecheap은 가입하려고 보면 대학생을 지원하는 지역이 미국, 호주, 영국(?)이라고 봤던 거 같다. 나는 미국 대학교 .edu 이메일이 있어서 문제가 없었지만, 문제가 된다면 Name.com을 시도해보라.

Azure VM

AWS 프리티어를 썼다가 1년이 끝나서 Azure로 갈아탔다. Azure for Students는 써본 결과 좀 까다로운 면이 있기는 하지만, 전반적으로 만족스럽다. 1년 무료이며 카드 연결 안해도 되고, $100 크레딧이 주어져서 유료 서비스라도 여기서 차감된다. 대신 무료 서비스라고 명시된 VM B1s라던가, Azure SQL DataBase 250GB S0는 하나하나 내가 꼼꼼하게 체크하며 만들어야한다. 하나라도 수틀리면 그냥 크레딧에서 차감된다.

여기를 참고해서 Nginx 설치까지 마치자. 처음에 VM 만들 때 인바운드 포트를 22(SSH)뿐만 아니라 80(HTTP), 443(HTTPS)까지 열어두자.

그리고 서버에 SSH로 접속해서 openssl로 Private Key를 생성해야한다.

$ openssl req -new -newkey rsa:2048 -nodes -keyout example.key -out example.csr

여기서 만들어진 파일들은 SSL 인증서를 만드는 데 쓰이고, Nginx에서 연결할 때 필요하기도 하다. File은 필요하다면 Filezilla SFTP 등으로 내 컴퓨터에 옮겨놓자.

Namecheap

깃허브 팩에서 Namecheap의 도메인뿐만 아니라 SSL 인증서까지 준다. SSL 인증서는 깃허브에서 주는 프로모션 코드를 복사해 사용하면 된다. Namecheap이 제공하는 도메인은 GitHub Pages 전용이라는 글을 봤던 기억이 있는데, 전혀 상관없다.

도메인 및 SSL 구매가 끝나면 Namecheap의 Domain list의 Manage 버튼을 클릭해 맨 위 Record처럼 Host의 값은 @으로 주고, Value는 Azure VM의 공용 IP 값을 입력해주면 된다. 적용되기까지 대략 30분 걸린다. 맨 아래 Record처럼 내가 구매한 도메인 이름은 저렇게 적혀있다. TTL 값은 기본이 30min이었는데 그냥 내가 바꿨다.

여기까지 마쳤으면 Azure VM에서 설치해두었던 Nginx 덕분에, 도메인에 접속하면 Welcome to Nginx가 뜰 것이다. 근데 SSL 인증서를 무료로 받아봤으면 써봐야하지 않겠는가? 443포트로, HTTPS로 접속할 수 있도록 설정해보겠다.

Nginx에 SSL 인증서 연결

SSL을 구입했으면 Namecheap에서 인증서 파일을 zip으로 다운받을 수 있다. Namecheap 말고 다른 사이트는 이메일로 날라온다던데 나는 직접 다운받았다.

다운받으면 ~.ca-bundle, ~.crt, ~.p7b 파일이 들어있다. 나는 그냥 crt 파일과 기존의 Private Key를 Nginx에 연결했다가 낭패를 봤는데, 이대로 사용하면 안된다. 여기 벨로그와 여기 공식문서를 참고했다. 파일을 묶어줘야 한다.

$ cat example.crt example.ca-bundle >> example_chain.crt

이렇게 해서 나온 crt파일을 Nginx에 연결하면 된다.

• 전체적으로 파티션 수를 늘려 병렬성 증가
• 매우 큰 파티션이나 Skew 파티션의 크기를 조절하려고
• 파티션을 분석 패턴에 맞게 재분배 (Write once, read many)
  • 어떤 DataFrame을 특정 컬럼 기준으로 그룹핑을 하거나 필터링을 자주 하는 경우 (미리 그 컬럼 기준으로 지정해두었다면 그게 Bucketing)

Coalesce가 필요한 경우

• 파티션의 수를 줄이는 용도 (늘리지 않음)
• 셔플링이 발생하지 않고 로컬 파티션들을 머지함 (따라서 Skew 파티션을 만들어낼 수 있음)
• Column이 사용되며 균등한 파티션 크기를 보장할 수 없음

Spark AQE (Adaptive Query Execution)

• Dynamic query optimization that happens in the middle of query execution based on runtime statistics
• Key points: 동적, 실시간 통계정보, 쿼리 중간에 바꿔준다

Dynamically coalescing shuffle partitions

• 왜 필요한가? (적당한 파티션의 크기와 수는 성능에 지대한 영향을 미침)
  • 너무 많은 수의 작은 파티션
    • 스케줄러 오버헤드
    • 태스크 준비 오버헤드
    • 비효율적인 I/O (파일시스템/네트워크)
  • 적은 수 큰 파티션
    • GC 악몽 (OOM=Out Of Memory)
    • Disk Spill
  • spark.sql.shuffle.partitions라는 하나의 변수로는 불충분

• 배치 처리 주기적으로 데이터를 한 곳에서 다른 곳으로 이동하거나 처리. 처리량(Throughput)이 중요

• 데이터 배치 처리

  • 처리 주기는 보통 분에서 시간, 일 단위
  • 데이터를 모아서 처리
  • 처리 시스템 구조
    • 분산 파일 시스템(HDFS, S3)
    • 분산 처리 시스템(MapReduce, Hive/Presto, Spark DataFrame, Spark SQL)
    • 처리 작업 스케줄링에 보통 Airflow 사용

• 데이터 실시간 처리

  Realtime	Semi-Realtime
  짧은 Latency	합리적인 Latency
  연속적인 데이터 스트림	배치와 유사한 처리(Micro-batch)
  이벤트 중심 아키텍처 (수신 데이터 이벤트에 의해 작업이나 계산이 트리거되는 구조)	적시성과 효율성 사이의 균형 (처리 용량과 리소스의 활용도를 높이기 위해 일부 즉각성을 희생하기도 함)
  동적 및 반응형 (데이터 스트림의 변화에 동적으로 대응하여 실시간 분석, 모니터링 및 의사 결정을 수행)	주기적인 업데이트

• 이벤트 데이터 모델 전송/저장

  • Point to Point Many to Many 방식. Throughput은 중요하지만 Latency가 중요한 시스템에서 사용 가능. 다수의 Consumer들이 존재하는 경우 데이터를 중복해서 보내야 함. Backpressure에 취약한 형태. 이 시스템의 경우에도 Consumer/Subscriber쪽에 작은 버퍼가 존재하지만 곧 부족해짐.
    • 🙄 Backpressure issue란? 스트리밍 시스템에서 데이터는 일반적으로 일정한 속도로 생성 (Producer). 하지만 가끔 데이터 생성이 폭발적으로 늘어날 수 있음. 다운스트림 단계(Consumer)에서 적시에 처리되어야 함. 하지만 들어오는 데이터 속도를 따라잡지 못하면, 시스템에 데이터가 쌓여 지연되면서 메모리 사용량 증가 등으로 잠재적인 시스템 장애 초래 가능
  • Messaging Queue Backpressure issue를 해결할 수 있는 방식. 그러나 완전하게 해결은 못함. Producer별로 Topic이 생성됨. 보통 micro-batch라는 형태로 아주 짧은 주기로 데이터를 모아서 처리하는데, Spark Streaming이 대표적임.

Day 3

• Kafka 실시간 데이터를 처리하기 위해 설계된 오픈소스 분산 스트리밍 플랫폼, 데이터 재생이 가능한 분산 커밋 로그 (Distributed Commit Log) Scalability와 Fault Tolerance를 제공하는 Publish-Subscription (= Producer-Consumer) 메시징 시스템 High Throughput과 Low Latency 실시간 데이터 처리에 맞게 구현됨 Retention Period 동안 메시지를 저장

• Kafka Broker = Kafka Server, Kafka Node

• Kafka Topic

  • Consumer가 데이터(Message)를 읽는다고 없어지지 않음
  • Consumer별로 어느 위치의 데이터를 읽고 있는지 위치 정보를 유지함
  • Fault Tolerance를 위해 이 정보는 중복 저장됨

• 직렬화 vs 역직렬화

  • 직렬화(Serialization): 데이터나 객체를 바이트로 변환한다는 의미. 객체의 상태를 저장하거나 전송할 수 있는 형태로 변환하는 프로세스. 보통 이 과정에서 데이터 압축 등을 수행. 가능하다면 데이터의 스키마 정보 추가
  • 역직렬화(Deserialization): Serialized된 데이터를 다시 사용할 수 있는 형태로 변환하는 Deserialization. 이 과정에서 데이터 압축을 해제하거나 스키마 정보 등이 있다면 데이터 포맷 검증도 수행

Day 4

• kafka-console-consumer 커맨드라인을 통해 Topic에서 Message 읽기 가능. --from-beginning 옵션이 있으면 처음부터 읽음(Earliest), 아니면 latest로 동작

• ksqlDB REST API나 ksql 클라이언트 툴을 사용해서 Topic을 테이블처럼 SQL로 조작

• Consumer Group

  • Consumer가 Topic을 읽기 시작하면 해당 Topic내 일부 Partition들이 자동으로 할당됨
  • Consumer의 수 < Partion의 수인 경우: Partition은 라운드 로빈 방식으로 Consumer들에게 할당됨 (하나의 Partition은 하나의 Consumer에게만 할당되므로)
  • 데이터 소비 병렬성 ↑, Backpressure ↓

• Consumer/Producer 패턴 많은 경우 Consumer는 한 Topic의 메시지를 소비해서 새로운 Topic을 만들기도 함. 즉 Consumer이면서 Producer로 동작

• 특강

  • 구직공고의 '이런 분이면 좋아요'는 가산점 느낌이 강하니까, 쫄지 말기
  • '3년 이상'이라는 경력이 적혀있다면, 2년까지는 지원해도 괜찮다고 보시는 편
  • 프로젝트 경험은 너무 기술적인 내용을 적기보다는, 어떤 점을 해결했고 그걸 가지고 뭘 했는지를 서술하는 것이 좋음.
  • 기술 질문에 답할 때 어떻게 해야할지 생각이 안난다면, 어떤 부분에서 막히는지 이야기 & 도움 요청 부끄러운 거 아님. 바로 답하지 않고 문제를 명확히 이해하려는 질문을 한다면 시간도 벌고 & 의사소통 열심히 하는 사람으로 보이고 일석이조

• Spark vs. MapReduce

  Spark	MapReduce
  기본적으로 메모리 기반, 다양한 컴퓨팅 지원, pandas 데이터프레임과 개념적으로 동일한 데이터 구조 지원	디스크 기반, 하둡(YARN) 위에서만 동작, key-value 기반 데이터 구조만 지원

Day 2

• Spark 데이터 처리 흐름

  • 데이터프레임은 작은 파티션들로 구성됨 (한 번 만들어지면 수정 불가)
  • 입력 데이터프레임을 원하는 결과 도출까지 다른 데이터프레임으로 계속 변환 (sort, group by, filter, join, ...)

• 셔플링: 파티션 간에 데이터 이동이 필요한 경우 발생

  • 발생하는 경우: 명시적 파티션을 새롭게 하는 경우 (파티션 수를 줄이기), 시스템에 의해 이뤄지는 셔플링 (그룹핑 등의 aggregation이나 sorting)
  • 데이터 처리에 병렬성을 주지만, Data Skewness가 발생하는 단점이 있기 때문에, 셔플링은 최소화하고 파티션 최적화를 하는 것이 중요함

• Spark DataFrame 실습

  df = spark.read.format("csv")\
       .option("inferSchema", "true")\
       .load("1800.csv")\
       .toDF("stationID", "date", "measure_type", "temperature", "_c4", "_c5", "_c6", "_c7")

  여기서 inferSchema 옵션은 원래 default가 False. 직역하면 스키마를 추론한다 → Spark가 해당 DataFrame을 만들 때 앞의 레코드들을 샘플링 해서 본 뒤 type이 무엇인지 추측하는 것임.

  from pyspark.sql.types import StringType, IntegerType, FloatType
  from pyspark.sql.types import StructType, StructField
  
      schema = StructType([ \
          StructField("stationID", StringType(), True), \
          StructField("date", IntegerType(), True), \
          StructField("measure_type", StringType(), True), \
          StructField("temperature", FloatType(), True)])

  앞 코드에서는 추론해보라고 했지만 이 코드는 명시적으로 알려주는 코드.

  • DataFrame의 컬럼을 지칭하는 방식 4가지 (모두 동일한 표현)

  from pyspark.sql.functions import col, column
  
  stationTemps = minTemps.select(
  "stationID",
  col("stationID"),
  column("stationID"),
  minTemps.stationID
  )

Day 3

• Window 함수: ROWS BETWEEN AND

  
  SELECT value FROM rows_test;
  
  SELECT
    SUM(value) OVER(
       order by value
       rows between 2 preceding and 2 following -- 뜻: 자기 기준 앞 2개 & 뒤 2개. 숫자 대신 unbounded도 가능
   ) AS rolling_sum
  FROM rows_test;

• Parquet: Spark의 기본 파일 format. 하나의 데이터 블록은 하나의 Row Group으로 구성됨

Day 4

• JOIN 2종류

  • 큰 데이터 - 큰 데이터 간의 JOIN: 보통 JOIN (Shuffle JOIN)
  • 큰 데이터 - 작은 데이터 간의 JOIN: Broadcast JOIN
    • spark.sql.autoBroadcastJoinThredshold 파라미터로 데이터프레임 하나가 충분히 작은지 여부 결정

• JOIN 할 때 생각해야 할 부분

  • Shuffle되는 양을 줄이는 방법(Filtering 먼저 하기, Grouping 먼저 하기 등)
  • 병렬화를 최대화할 방법 찾기(Executor 수, Partition 수)
  • JOIN 키 값의 분포 확인 후 균일하게 만들 방법 찾기 (Data Skew)
  • Partition 로딩 시에 버킷팅을 통해 키가 같은 레코드들이 같은 파티션에 존재하도록 설정

• UDF (User Defined Function) DataFrame이나 SQL에서 적용할 수 있는 사용자 정의 함수

docker-compose.yaml

# 이렇게 설정해두면 웹 UI에는 안 보임. 하지만 웹 UI에서 세팅하는 거랑 똑같은 효과
environment:
    AIRFLOW_VAR_DATA_DIR: ~ # AIRFLOW_VAR로 시작하면 Airflow의 환경변수
    AIRFLOW_CONN_BLAH_BLAH: ~ # AIRFLOW_CONN으로 시작하면 Airflow 웹 UI에 세팅된 connection이랑 똑같음, 그 뒤의 string이 connection의 이름이 됨.

.airflowignore

• Airflow의 DAG 스캔 패턴: dags_folder가 가리키는 폴더를 서브폴더 포함 모두 스캔해서 DAG 모듈이 포함된 모든 파이썬 스크립트를 실행함 → 가끔 사고로 이어짐
• .airflowignore는 Airflow가 의도적으로 무시해야 하는 디렉토리/파일 지정
• 예시 (정규표현식)
  • .airflowignore 안의 내용이 아래와 같다면
    • project_a → project_a_dag_1.py, TESTING_project_a.py, project_a/dag_1.py 이와 같은 파일 무시
    • tenant_[\d] → tenant_1.py 이와 같은 파일 무시

✔ 6/20

Airflow API 활성화

• airflow.cfg의 api section 내용

  [api]
  auth_backend = airflow.api.auth.backend.basic_auth

• docker-compose.yaml AIRFLOW__API__AUTH_BACKENDS: 'airflow.api.auth.backend.basic_auth,airflow.api.auth.backend.session 여기에서 처음 언더바 2개 다음의 글자(API)는 section을 의미, 그 다음 언더바 2개 다음 글자(AUTH_BACKENDS)는 section 밑의 key를 의미.

• is_active와 is_paused 의미 curl -X GET --user "airflow:airflow" http://localhost:8080/api/v1/dags 이와 같이 GET 요청을 했을 때 dags에 관한 정보들이 뜸. 여기서 is_active는 dags folder에 존재하는지 여부이고, is_paused는 현재 멈춰 있는지 (활성화된 dag인지 아닌지) 확인할 수 있는 값.

✔ 6/21

Airflow API - variables

환경변수로 지정된 건 return하지 않음. 예를 들어 DATA_DIR이라는 환경 변수 존재 확인하려면 $ docker exec -it {컨테이너_ID} airflow variables get DATA_DIR -> 결과 출력됨.

Jinja Template

• bash_command (str), env (dict[str, str])는 Jinja Template 지원. 여기 참고해서, 파라미터 설명에 (templated)라고 되어있으면 Jinja Template 지원한다는 의미

Sensor

• 특정 조건이 충족될 때까지 대기하는 Operator
• Airflow의 내장 Sensor들
  • FileSensor: 지정된 위치에 파일이 생길 때까지 대기
  • HttpSensor: HTTP 요청을 수행하고 지정된 응답이 대기
  • SqlSensor: SQL DB에서 특정 조건을 충족할 때까지 대기
  • TimeSensor: 특정 시간에 도달할 때까지 워크플로우를 일시 중지
  • ExternalTaskSensor: 다른 Airflow DAG의 특정 작업 완료를 대기, 실수하기 쉬우므로 딱히 쓰지 않는 걸 추천하셨다.
• 기본 모드가 poke고, worker 하나를 붙잡고 거기서 계속 체크하고, 기다렸다가 또 체크하고... → worker 하나가 낭비되는 특징

DataGrip 스키마 안 보일 때

• 여기 참고했다. 저번에 팀 프로젝트 할 때도 안 보여서 어쩌다보니 해결했는데.. 실습할 때도 안 보이는 스키마가 있었다. 새로 뭘 하기 전에 앞으로는 미리 확인 해놓아야지

Airflow 메타데이터 DB 내용 확인

1. $ docker exec -it {CONTAINER_ID} sh
2. $ psql -h postgres
3. $ \dt
4. 하고 싶은 쿼리 날리기

✔ 6/22 ~ 6/23

DBT - Sources

• 기본적으로 처음 입력되는 ETL 테이블을 대상으로 함
  • 별칭(alias) 제공
  • 최신 레코드 체크 기능 제공

Airflow dags 실행 안되는 문제 해결

터미널에서 airflow dags test로 하면 DAG 실행이 잘 되지만, localhost:8080에 들어가서 dag를 trigger하면 실행이 안되는 문제가 발생했다.

• airflow-worker에서는 PermissionError: [Errno 1] Operation not permitted: ~~~가 떴다.

• airflow-scheduler에는 Executor reports task instance <TaskInstance: {DAG_ID}.extract scheduled__2023-06-21T00:00:00+00:00 [queued]> finished (failed) although the task says its queued. (Info: None) Was the task killed externally? 가 떴다. 그것도 모든 DAG가... 나한테 왜 이래. airflow-worker에서 알려준 경로에 log 파일이 존재해야 하는데 쓰기 권한이 있는데도 존재하지도 않았다. DAG는 trigger한 이후로 계속 running 상태 또는 다 fail.. 그래서 나름대로 해결해보려고 삽질을 일주일 내내 했다. 😫

• 원인? 일단 예상되는 원인은 airflow 2.5.1 자체에 버그가 있다는 것. 구글링 하다가, 간신히 나랑 똑같은 오류가 발생했다는 글을 찾았다. 나는 airflow 2.5.1 버전을 docker-compose해서 쓰고 있었는데 글쓴이의 버전도 나랑 똑같다. bug를 고쳐서 커밋하셨길래, 봤더니 try-except 구문으로 처리된 코드가 추가되어 있었다.

  그래서 당장 내 환경에서 해당 부분을 확인해봤다. 경로는 docker container에 접속해서 여기로 이동했다. → /home/airflow/.local/lib/python3.7/site-packages/airflow/utils/log/file_task_handler.py

  

  음.. 나에겐 그런 부분이 존재하지 않아 오류가 발생했던 것 같다. 물론 나도 확실히 모름 ㅎ 근데 여기서 내가 고치는 게 말도 안될 것 같아서 그냥 airflow 2.6.1 버전 docker-compose.yaml파일 새로 받고 다시 했다. 2.6.1 버전 확인해보니 해당 부분이 try-except로 처리되어 있었다. DAG 실행도 다시 잘 됐다. 👍

dbt Version 업그레이드 하기

pip3로 install하면 dbt 버전이 1.0.0 보다 한참 낮은 버전으로 설치되는 문제가 발생했다. 얘는 seeds 폴더를 못 알아먹는다,, 실습해야 하는데 마음에 안 든다. dbt가 1.0.0 버전 이후로는 pip3 install upgrade 조차 지원을 안한다고 한다. 그럼 어떻게 해결했느냐?

1. GitHub에서 가장 최근 버전인 dbt-core-1.5.2.tar.gz를 다운로드 받자.
2. 저장하고 해당 파일 있는 곳에서 압축 풀자. $ tar -xvf dbt-core-1.5.2.tar.gz
3. 안으로 들어가자. $ cd dbt-core-1.5.2
4. 설치하자. $ python setup.py install
5. 버전 확인하자. $ dbt --version
6. 그럼 이제 postgres 또는 redshift 플러그인을 설치할 차례. 각자 하고 싶은 거 깔자. $ pip3 install dbt-redshift 그러면 core 버전에 맞게 plugin도 높은 버전으로 깔린다. 아래처럼!

ENTRYPOINT vs. CMD

1. ENTRYPOINT가 있으면 CMD 값이 파라미터로 실행됨
2. 아니면 CMD가 실행됨

docker run vs. docker exec

• 공통점: 두 명령 모두 Container ID 필요, --user root 또는 -u root를 통해 루트 유저로 연결 가능
• docker run: 새로운 Container를 실행
• docker exec: 이미 실행된 Container에 작업하는 것

✔ 6/13

Docker 컨테이너 내부 프로세스가 오픈한 포트번호를 외부로 노출시키기

• 포트맵핑(=포트포워딩) 예시로 내부에 4000 포트 열었고, 외부에도 4000번 포트로 오픈하고 싶다면,$ docker run -p 4000:4000 {이미지_이름}

✔ 6/14

Docker Volume

• Docker Container 내의 가상 파일 시스템과 호스트 시스템 파일 시스템을 맵핑

  • 예시) 호스트 파일 시스템의 /home/heering/logs를 Docker Container의 /var/lib/airflow/logs로 맵핑
  • 이 경우 Docker Container가 중단되더라도 모든 Airflow logs는 기록이 남게 됨

• Dockerfile

  • VOLUME 명령을 통해 anonymous volume만 지정 가능

• docker-compose

  • Host Volume이나 Named Volume을 사용하는 것이 일반적

✔ 6/15

Docker-Compose

• 다수의 컨테이너로 소프트웨어가 구성되는 경우 사용할 수 있는 툴 + 환경설정 파일

  • docker-compose.yml로 설정

• 개별 컨테이너를 따로 관리하는 것보다 훨씬 더 생산성이 높음

• docker-compose down: 컨테이너 정지 + 삭제, 이미지는 삭제 안함

• docker-compose.yml의 depends_on

  • condition으로 가능한 값

  service_started	service_healthy	service_completed_successfully
  service가 healthy한지 아닌지 상관없이 해당 컨테이너가 시작했으면 나를 실행해라.	해당 service가 healthy할 때 내 서비스 실행	초기화 작업 한 번 해주고, 초기화 작업이 성공적으로 끝나면 뒤에 있는 컨테이너들이 실행하도록 해야 할 때가 있음. 초기화 해야하는 컨테이너 제외, 나머지 모든 컨테이너들이 depends_on: {초기화 해야 하는 서비스명}: condition: service_completed_successfully.

Docker-Compose build 안될 때 해결법

$ docker-compose -f docker-compose.mac.yml build 를 하려 했더니, 아래처럼...

failed to do request: head "https://registry-1.docker.io/v2/library/python/manifests/3.9-slim": proxyconnect tcp: ~~ 매우 긴 오류

오늘도 어김없이 나타난 오류. 전체 다 복사해서 검색했더니 안 나와서 영어로 대충 지어내니까, 또 stack overflow에 해답이 있었다. 😆

Windows 10 + wsl2 환경에서, 아래처럼 입력하고 sudo vi /mnt/c/users/<username>/.docker/config.json

수정을 아래와 같이 한다. credsStore에서 s 하나 빼기. 근데 사실 이거는 눈속임이고 json파일 자체를 지워도 되지 않을까 싶다..?

{
  "credStore": "desktop"
}

이렇게 해도 오류가 난다면 Docker Desktop restart를 해볼 것.

• MySQL or PosgreSQL 테이블이라면 만족해야하는 것들
  • created (timestamp): Optional
  • modified (timestamp)
  • deleted (boolean): 레코드를 삭제하지 않고 deleted를 True로 설정

Backfill을 커맨드라인에서 실행하고 싶다면

$ airflow dags backfill dag_id -s 2023-01-01 -e 2023-06-09

• 단, catchUp이 True, execution_date를 사용해서 Incremental update가 구현된 상태
• start_date부터 시작, end_date는 미포함
• 실행 순서는 랜덤. 날짜 순으로 하려면 default_args = {'depends_on_past': True, ... }

Airflow 정리

• 파이썬으로 작성된 데이터 파이프라인(ETL) Framework
  • Airflow에서 데이터 파이프라인이 DAG임. (Directed Acyclic Graph)
• 장점
  • 데이터 파이프라인 세밀하게 제어 가능
  • Backfill이 쉬움
• 스케일링 방식: Scale up vs Scale out vs Cloud version vs K8s

여기에서 기록했던 것처럼, 빅데이터 웨어하우스에서는 PK가 무시된다. 오늘은 PK를 유지하는 나름의 방법을 배웠다. 아래는 예시다.

1. 임시 테이블을 만든다. 원래 테이블의 내용을 임시 테이블 t에 복사한다.

CREATE TEMP TABLE t AS SELECT * FROM heering.weather_forecast;

2. DAG는 임시 테이블(staging table)에 레코드를 추가한다. 중복 데이터가 들어갈 수 있다.

3. 원래 테이블 내용 삭제한다.

DELETE FROM heering.weather_forecast;

4. 중복을 없앤 형태로 새로운 테이블을 생성한다.

INSERT INTO heering.weather_forecast
SELECT date, temp, min_temp, max_temp, created_date
FROM (
    SELECT *, ROW_NUMBER() OVER (PARTITION BY date ORDER BY created_date DESC) seq
    FROM t
)
WHERE seq = 1; -- 일련번호인 seq가 1인 것만.

5. 매번 이렇게 새로 덮어쓰는 형식의 업데이트. autocommit=True라면, 최소 3번과 4번은 transaction으로 처리되어야 함.

Backfill 관련 Airflow 변수

from airflow.decorators import task
from airflow import DAG

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

@task
def print_goodbye():
# ...

with DAG(
# ...
) as dag:

print_hello() >> print_goodbye() # 함수 이름이 task ID가 됨

Docker에 파이썬 모듈 설치하기

내가 까먹을까봐 적어두는 글

1. $ docker ps 하면 목록 다 뜸
2. $ docker exec -it {CONTAINER_ID} sh 원하는 컨테이너 ID 입력해서 shell script 실행하고
3. $ pip3 install yfinance 거기에서 예를 들어 이렇게 설치

[참고] root user로 로그인 하는 방법: $ docker exec --user root -it {CONTAINER_ID} sh

기타

dags 폴더에서 코딩할 때 주의할 점

• Airflow는 dags 폴더를 주기적으로 스캔
  • DAG 모듈이 들어있는 모든 파일들의 main 함수가 실행이 됨 → 의도하지 않았지만, 테스트 코드까지 실행될 수 있음

Time Zone

타임존은 UTC 권장

try:
    cur.execute(create_sql)
    cur.execute("COMMIT;")
except Exception as e:
    cur.execute("ROLLBACK;")
    raise # 이렇게 하기!!!

Docker Desktop - Airflow 환경 세팅하기

나는 Windows인데 강의에서는 Mac으로 설명해주시는 안타까운 상황..

혹시나 했는데 역시나!! 강의에서는 순조롭게 진행되고 있지만, 내 컴퓨터에서는 문제가 발생했다. 왜 내껀 안 뜨는 거야 😕

오늘도 stackoverflow의 도움을 받아.. Powershell에서 이 명령어 입력. wsl --set-version {Distro_Name} 2 참고로 엄청 오래걸린다. backspace 눌러서 완료됐나 확인할 것.

메모리 할당을 wsl2에 6GB로 해뒀는데, 미적용 된 것 같아서 찾아보니 해결은 여기를 참고해서 했다. 무조건 powershell로..

문제 발생 당일. 당시 리드미를 확인했을 때 내 계정이 로그인 되어있었기에 문제 없어 보였지만, 다른 컴퓨터로 들어가봤더니 내가 올렸던 이미지가 깨져있었다. 🫠

원인이 뭘까하고 링크를 보니까 예전에는 링크가 user-content~~~ 이런 식으로 시작했던 것 같은데, 내_레포_이름~~~ 형식으로 바뀌어있었다. 만약 내가 이미지 링크를 받아왔던 repository가 public이라면 다른 사용자는 이미지가 보이지만, private라면 문제가 된다.

찾아보니 이는 깃허브가 5월 9일부터 적용한 정책 방식 때문이었다. 공식 블로그에도 올라왔던데 이제 알았다. 앞으로는 이미지 올릴 해당 repository 이슈에서 작업해야겠다.

• 참고1

• 참고2

Snowflake 특징

• 가변 비용 모델
• csv, json, avro, parquet 등 포맷 지원
• S3, GC 클라우드 스토리지, Azure Blob Storage 지원
• Time Travel (과거 데이터 쿼리 기능으로 트렌드 분석 쉽게)
• 웹 콘솔 말고도 Python API를 통해 관리/제어
• 클라우드 스토리지를 외부 테이블로 사용 가능
• Data Sharing ("Share, Don't move"): Copy 개념이 아닌, 데이터셋을 사내 혹은 파트너에게 스토리지 레벨에서 공유하는 방식
• Redshift와 달리 Group 지원하지 않음. Role은 계승 가능

Tip

AWS 어드민 사용자의 AWS KEY ID, SECRET KEY를 사용하지 말고 새로 계정 하나 만들어서 권한을 주자. 어드민 사용자 쓰다가 실수로 깃허브에 노출이라도 되면...

[파트 07]

Cohort(코호트)

• 특정 속성을 바탕으로 나눠진 사용자 그룹 (보통 속성은 사용자의 서비스 등록월)

Cohort 분석

• 코호트를 기반으로 사용자의 이탈률, 잔존율, 총 소비 금액 등을 계산

Cohort 기반 사용자 잔존율 (Retention)

• 보통 월 기반으로 시각화해서 보는 것이 일반적