• 이미지
• 테이블 형식(csv, parquet, DB)
• 텍스트
• 동영상

2. Process 2-1. 데이터 수집(Collect) Classification이 목적이라면 레이블 별로 최소/최대 레이블의 비율이 1:10은 되어야함 1: 100 이래 버리면 분류 힘들다는.. 2-2. 데이터 준비(Inspect format or label) 2-3. 학습 (set parameters and build/training) Training/ Validation/ Test 8 : 1 : 1 비율 2-4. 평가 (estimation) 2-5. 배포 및 예측(deploy and predict/inference)

3. Vertex AI Guide 3-1. Managed Training(관리형 학습)

• 학습 코드 작성
• code submit
• code monitoring
• 클라우드니까 머신타입 당연히 골라야 하고 GPU타입(Accelerator)도 고르네요.. 관리형이라면서 다 해야하네.. ㅎ 서버리스랑은 약간 다른개념^^.

  아래와 같은 예제들이 있으니 어떤식으로 하는지 상상해보기!

  

3-2. Managed Inference(관리형 추론)

• 3-1 에서 만든 모델을 Vertex AI Model Registry에 업로드
• 아마 Google Cloud Storage에 있지 않을까 싶네요
• 아래와 같이 수행하면 추론서버(RESTful Inference EndPoint)로 사용 가능

주로 오픈소스 만들 때, 클래스 상속 시 특정 메소드 필수 구현 강요 "Cat 클래스는 speak 메소드를 정의 안해서 에러 발생" 이 부분이 제일 중요^^.

from abc import ABC, abstractmethod

class AbstractAnimal(ABC):

    @abstractmethod
    def speak(self):
        pass

class Dog(AbstractAnimal):
    def speak(self):
        return "Woof!"

class Cat(AbstractAnimal):
    def speak_test(self):
        return "Meow!"

# 추상 클래스를 상속받은 구체 클래스의 인스턴스 생성
dog = Dog()
print(dog.speak()) # 출력: Woof!

cat = Cat()
print(cat.speak_test()) # 출력: TypeError: Can't instantiate abstract class Dog with abstract method speak!

# 추상 클래스 직접 인스턴스화 시도 (오류 발생)
try:
    animal = AbstractAnimal()
except TypeError as e:
    print(e) # Cannot instantiate abstract class AbstractAnimal with abstract method speak

from django.urls import reverse
from django.test import TestCase

class MangangGetTest(TestCase):
    @class
    def setUpTestData(cls):
        cls.mock_attr1 = ''
        cls.mock_attr2 = ''
        cls.mock_url = '/path'
    def test_mangang_post(self):
        mocked_data = {
                    mock_key_1: '모킹 하세요1',
                    mock_key_2: '모킹 하세요2',
                }
        # 각각 GET, POST에 대한 검사를 하기 위한..
        get_GET_response = self.client.get(mock_url)
        get_POST_response = self.client.post(reverse('aliasing_view', mocked_data))

각각 django TDD에서 필수적이다

1. reverse는 url에 aliasing 해놓은 비즈니스 로직 호출
2. TestCase는 class에 상속해서 테스트 코드 구성 (CBV 형식)

setUp : 하위의 모든 테스트 코드 호출 시에 매번 셋업하는데 함수 많으면 오래 걸려서 아래 setUpTestData 사용이 유리한 상황이 있음 setUpTestData: @classmethod와 같이 사용 cls.attr1 = value 선언 후, 하위 함수에서 self.attr1 호출 가능 ** 모든 함수의 시작은 test로 시작해야함 def test_check_func1(self)..

• 운영 체제에서 관리하고 있다. 호스트 내의 파일 경로에서 확인은 어렵고 가상 메모리의 아래 경로에 있다. /proc/meminfo /proc/stat /proc/net/dev /prod/diskstats

시스템 메트릭은 가상 메모리에서 실시간으로 위 경로에 write 되고 있다! 프로메테우스의 node exporter는 이러한 시스템 메트릭을 외부 포트로 노출 하고(default:9010) 프로메테우스 클라이언트 서버에서 타겟 서버의 9010포트에서 특정 interval단위로 스크랩 해가는 것이다!

global: scrape_interval: 15s evaluation_interval: 15s

scrape_configs:

• job_name: 'prometheus' static_configs:
  • targets: ['localhost:9090']

ex) target_server_host: http://localhost:9090 client_server는 target_server에서 interval: 15s 단위로 시스템 메트릭을 수집하고 있다.

DPU의 Pricing

• Data Processint Unit 의 약자임 (젠슨 황-엔비디아 대표 언급)
  • 1 DPU = 4 vCPU & 16 GB of memory ( $0.44 per Hour )
• Data Catalog
  • $1.00 per 100,000 objects per Month
• Requests
  • 1M(1,000,000) 은 무료
  • $1.00 per 1M requests above 1M in a month

WorkFlow

csv -> Transform -> Parquet

1. Bucket 생성( S3 )
2. IAM 권한 부여(적절하게..)
3. Glue 내에 DB 생성
4. Crawler 생성 후 실행
5. ETL Job 생성 후 실행(Extract, Transtorm and loads)

Checklist

Job Bookmark 확인 DB Procedure ( DW -> DM ) python( Landing Zone -> DW 로 CUD 작업 ) 랜딩존의 이미지 메타데이터(URL) -> 웹URL ETL 과정을 모니터링(Log)

Process

1. 버킷에 iris.csv 테스트 데이터 생성 (s3://kimjm/test-csv-/iris.csv)
2. database 생성( kimjm_test_postresql)
3. crawler 생성( Data Sources 지정 -> 버킷 경로 )

Test WF(Work Flow)

trigger - node - trigger(any/all) -> node

1. trigger ( 스케쥴, on-demand )
2. forward 2-1. treigger( depends on 2 )

Data Lake Overall Flow

1. ERP서버(REST API)에서 S3로 일단위로 추출
   • 증분(incremental), 증분테이블(CDC) or 타임스탬프 컬럼 기반
     • original.parquet = original.parquet + added.parquet(add는 날짜)
   • 모든 데이터 다 갖고 오기 (비효율적일듯)
     • original.parquet = original.parquet
2. S3 to Data-Catalog
   • crawler 실행( bookmark: enabled)
3. Data-Catalog( CDC만 있는)에 있는 증분 데이터를 DW에 반영
   • working_df = SparkSession.read(Data-Catalog)
   • original_df = SparkSession.read(Database) 두 df를 join해서 없는 것은 filter로 drop하고,

- 클라우드 환경에서의 ETL

내용이 어렵고 와닿지 않으실 수 있습니다. 요약하자면 이렇습니다. DB는 24시 존재하고, 어떠한 애플리케이션(서비스)에도 존재합니다 이제 분석이 필요합니다 하지만 DB에 접근해서 분석하지 않습니다. 음식의 맛, 성분 분석을 식당에서 하지 않고, 포장해와서 집에서 분석한다고 생각하시면 됩니다(먹방 리뷰 유튜버처럼 ㅎㅎ)

저장되는 DB뿐만이 아니라 데이터가 발생할 수 있는 모든 곳에서 스토리지로 데이터를 우겨넣습니다!!(실시간으로 야금야금이든.. 일괄로 뭉태기로 넣든!) 초기 구축 시에는 일괄로 넣어야 하니 아래의 방법중 2번으로 처음에 넣어도 좋겠네요.

Source - ETL툴 - Destination(A-B-C)구조입니다. B가 A에서 C로 갖다놓는데, 상황이 많기에 상황에 따른 적절한 전략을 짜야합니다. 전략이라면 데이터 유실없이 최적의 비용으로 유지보수가 쉬워야겠죠?????(젤 어려운말..) 운영계 DB에서 DW,DL 같은 스토리지 환경으로 가져갈 때 전략

1. Serverless(AWS Lambda) : 변화에 대한 초단위 배치
2. 인스턴스 스냅샷 : 이건 1회성
3. CDC(Change Data Capture): 변화 데이터에 대해서 read 하고, 목적지에 write(put)

AWS DMS(Database Migration Service) Source와 Destintation 사이에 인터페이스 호환성이 뛰어납니다 Source가 어떤 스택이든.. Destination이 어떤 스택이든..

여러 개의 스레드가 파이썬 바이트코드를 한번에 하나만 사용할 수 있게 락을 거는 것을 의미한다. 쉽게 말해서 하나의 스레드만 파이썬 인터프리터를 제어할 수 있도록 하는 뮤텍스라고 보면 된다.

For The MEMORY !

GIL의 설계배경을 한마디로 말하자면 메모리 안전성을 보장하기 위함이다.

그림에서 보듯이 우측의 객체는 참조의 수가 2인데, 좌측의 객체는 참조가 없어져 0이 된다. 이 개수가 0에 도달하면 개체가 점유한 메모리가 메모리 가비지 컬렉터에 의해 해제된다.

```python

reference count

import sys x = [] print(sys.getrefcount(x)) # x의 참조 카운트를 가져옵니다. (기본적으로 2입니다)

y = x print(sys.getrefcount(x)) # y가 x를 참조하므로 카운트가 1 증가합니다.

del y # y의 참조를 제거합니다. print(sys.getrefcount(x)) # 카운트가 다시 2로 감소합니다.

Garbage Collection 및 순환 참조 예제

import gc

def create_cycle(): x, y = {}, {} x['y'], y['x'] = y, x # x와 y는 서로 참초하므로 순환 참조 발생

create_cycle() gc.collect() print(gc.garbage)

if not reference_count:

문제는 이 레퍼런스 카운팅 변수가 멀티 스레드 환경에서 두 스레드가 동시에 값을 늘리거나 줄이는 Race Condition이 발생할 수 있다는 것이다. 이러한 상황이 발생하면 메모리 누수가 발생하거나 객체에 대한 참조가 남아있는 데도 메모리를 잘못 해제할 수 있다.

IPv4 대역 일부를 이용 사설망, Private Network 24비트 블록 10.0.0.0 ~ 10.255.255.255 -> 2^24개 00001010 00000000 00000000 00000000 20비트 블록 172.16.0.0 10101100 00010000 00000000 00000000 16비트 블록 192.168.0.0 11000000 10101000 00000000 00000000

VPN 2가지 종류

IPsec VPN Layer3 Site to Site (본사와 자사)

SSL VPN Layer4 Client to Site (사용자가 어느 장소에서든 접근 가능)

IPSec(Internet Protocol Security)

운용 모드

• Transport Mode:단말 to 단말

• Tunnel Mode: VPN to VPN

     Origin IP Header(사설 IP)를 캡슐화하고 IP 패킷 전체를 인증 혹은 보호하며, VPN 장비가 이를 도맡아 하므로 사설 네트워크와 사설 네트워크 간 통신을 가능케 합니다.

L2,L3, L4, L7

상위 계층을 활용할 수 있는 장비들은 모두 하위 계층 또한 이해하고 활용할 줄 알아야 한다. "L4 Virtual Server는 TCP/UDP만을 다루는 로드밸런서이고, L7 Virtual Server는 주로 HTTP/HTTPS만을 다루는 로드밸런서 아닌가요?" <Layer 4와 Layer 7(출처 : https://www.freeism.co.kr/wp/archives/698)> 상위 레벨의 프로토콜 이해 없이 TCP/UDP 로드밸런싱이 필요하면 L4 Virtual Server를 사용하면 됩니다. HTTP를 비롯한 Layer 7 헤더 해석이 필요하다면 L7 Virtual Server를 사용하면 되겠죠.

• layer7 헤더해석이 필요없는 트래픽은 layer4에서 밸런싱 하고 끝
  • HTTP는 TCP 기반의 프로토콜입니다. 그렇기 때문에 HTTP 통신을 하기 위해서는 반드시 3-way handshake를 실시하여 신뢰성 있는 연결을 생성하여야 합니다. 그다음에 HTTP GET을 통해 리소스를 얻어오거나 POST를 통해 업데이트를 실시하는 것입니다.

L4 Load Balancing

• IP와 Port를 활용하여 서버부하분산

  L7 Load Balancing

• IP와 Port를 사용하여 로드밸런싱을 하는 것은 같으나 Layer 7 프로토콜을 통해 사용자 정의 로드밸런싱을 실시하거나 Layer 7 프로토콜 헤더를 조작 / 활용할 수 있다는 특징이 있습니다.
• DNS 는 TCP/ UDP 모두 사용. 기본적으로 UDP 이지만, 전달해야 하는 패킷의 크기가 512bit 를 넘는다면 TCP를 사용, DNS를 제대로 Load Balancing 하기 위해서는 TCP/UDP Virtual Server 모두 생성해야함.

Client - Web Server - BackEnd

• Client와 Nginx 사이에는 https
• Nginx와 django(web framework) 사이에는 http
• 성능: HTTPS는 암호화와 인증 과정이 필요하기 때문에 CPU 사용량이 높을 수 있습니다. 클라이언트와 Nginx 사이에서만 HTTPS를 사용하면, 이 오버헤드를 한 번만 처리하면 되어 백엔드에서의 추가 부하를 피할 수 있습니다.
• 단순화: HTTPS 인증서 관리와 SSL/TLS 설정은 복잡할 수 있습니다. 이러한 설정을 Nginx에서만 처리하면 백엔드 설정이 더 간단해집니다.
• 보안: Nginx와 Django가 같은 호스트 또는 동일한 내부 네트워크 내에서 실행되고 있다면, 내부 트래픽은 이미 보안된 환경 내에 있으므로 HTTPS로 암호화할 필요가 없을 수 있습니다.
• Nginx 설정에서 이런 구성을 사용할 때, proxy_pass 지시어는 http로 시작하는 URL을 대상으로 합니다. 예를 들면:

  nginx
  Copy code
  location / {
    proxy_pass http://127.0.0.1:8001;

  ... (나머지 proxy_set_header 설정들)

  }

API_KEY=$PINECONE_API

Region

environment='gcp-starter'

import pinecone

API_KEY = 'your_api_key'
ENVIRONMENT = 'your environment'

#클라이언트 생성
pinecone.init(api_key=API_KEY, environment=ENVIRONMENT)

if index not in pinecone.list_indexes():
    index = pinecone.create_index('tutorial', dimension=1536, metric='cosine')

index = pinecone.Index('tutorial')

df = pd.DataFrame(data={'question':question_list, 'vector':vector_list})

index.upsert(vectors=zip(df.question, df.vector))
벡터를 생성하는 데 사용할 임베딩 모델은 text-embedding-ada-0021536차원의 벡터를 출력합니다. ~~텍스트~~

진짜 별거 아닌데,

docker ps, docker-compose up 같은 명령어 sudo 안주면 permission denied 준다

``` #sudo user modify -add -Group docker 사용자이름 #수퍼유저 권한으로 사용자를 도커 그룹에 추가 하겠습니다. sudo usermod -aG docker $USER

#새로고침 newgrp docker

VectorDB란..?

Large Language Model (LLM)의 솔루션이다. 기존 스칼라 기반 데이터베이스에서 임베딩 작업의 어려움을 해결

1. Vector Embedding (벡터로 표현됨) = [ 0.34, 0,12 , -0.03, 1.3 ]
2. Vector Embedding과 원본 콘텐츠와 같이 Vector DB에 Insertion
3. Apllication이 Query를 발행 하여 Embedding Model을 통해 Embedding을 생성하고, 해당 임베딩을 사용하여 Vector DB에서 유사한 Vector Embedding을 쿼리 (한 마디로 벡터값 비슷한 문장 갖고 온다는 뜻)

   Vector DB 장점

4. 데이터 관리: 삽입 삭제 업데이트와 벡터 인덱스를 사용하는 것보다 벡터 데이터를 쉽게 관리하고 유지할 수 있다.
5. 메타데이터 저장 및 필터링: 벡터 항목 관련 메타데이터를 저장할 수 있다. 필터를 이용하여 DB쿼리 가능
6. 확장성:데이터 양이 많아질 수록 분산 및 병렬 처리됨 ( 샤딩, 수평 분할 )
7. 실시간 업데이트: 실시간 데이터 업데이트를 지원( 동적으로 변경 가능 )
8. 백업 및 수집: 모든 데이터를 백업하는 일상적인 작업 처리(배치), Collections 형태로 백업 가능(NoSQL 특성)
9. 에코시스템 통합: Spark, 태블로, 그라파나 같은?? 분석, 시각화 툴들과 생태계 연동 가능

   작동 메커니즘

   벡터 데이터베이스는 벡터에서 작동한다 -> 기존 디비에서 최적화, 쿼리 날리던 방식과는 다르다는 의미 Similarity Metrics 를 사용 ( 벡터 유사도를 보는 거겠죠 ?? )

-> 가장 유사한 벡터를 찾기 벡터 데이터베이스는 ANN(Approximate Nearest Neighbor) 검색에 모두 참여하는 서로 다른 알고리즘의 조합을 사용한다. -> 해싱, 양자화, 그래프 기반 검색 대략적인 결과를 제공하기 때문에 제일 중요한 트레이드 오프(Trade-Off) 는 바로 ..!! -> 정확도, 속도 이겠죠?? 높은 정확도, 낮은 속도 낮은 정확도, 높은 속도 좋은 시스템은 높은 정확도 높은 속도(비용이 많이 들겠죠) 를 제공합니다(PineCone)

1.Indexing

2.Query

3.Post Processing

알고리즘

1. Random Projection( 무작위 투영 )

2. Product Quantization

3. Locality-sensitive hashing

4.Hierarchical Navigable Small World(HNSW)

유사성 측정

코사인 유사성:

벡터 공간에서 두 벡터 간의 각도의 코사인을 측정합니다. 범위는 -1에서 1까지이며, 여기서 1은 동일한 벡터를 나타내고, 0은 직교 벡터를 나타내고, -1은 정반대의 벡터를 나타냅니다.

유클리드 거리:

벡터 공간에서 두 벡터 사이의 직선 거리를 측정합니다. 범위는 0에서 무한대까지이며, 여기서 0은 동일한 벡터를 나타내고 값이 클수록 점점 더 다른 벡터를 나타냅니다.

내적:

두 벡터 크기의 곱과 두 벡터 사이 각도의 코사인 값을 측정합니다. 범위는 -∞에서 ∞까지이며 양수 값은 같은 방향을 가리키는 벡터를 나타내고 0은 직교 벡터를 나타내고 음수 값은 반대 방향을 가리키는 벡터를 나타냅니다.

필터링

사전 필터링:

이 접근 방식에서는 벡터 검색 전에 메타데이터 필터링이 수행됩니다. 이렇게 하면 검색 공간을 줄이는 데 도움이 되지만 시스템에서 메타데이터 필터 기준과 일치하지 않는 관련 결과를 간과할 수도 있습니다. 또한 광범위한 메타데이터 필터링으로 인해 계산 오버헤드가 추가되어 쿼리 프로세스가 느려질 수 있습니다.

사후 필터링:

이 접근 방식에서는 벡터 검색 후에 메타데이터 필터링이 수행됩니다. 이렇게 하면 모든 관련 결과를 고려하는 데 도움이 될 수 있지만 검색이 완료된 후 관련 없는 결과를 필터링해야 하므로 추가 오버헤드가 발생하고 쿼리 프로세스 속도가 느려질 수도 있습니다.

ELK를 Docker-Compose 로 구성할때 발생한 에러

ElasticSearch 오피셜 git repo를 클론 떠서 compose up 했을 때에 credentials 에러가 발생 그래서 찾아본 결과 docker-hub 회원가입하고 CLI 에 docker-login 해야함 그래야 Credentials 인증 되는데 cd ~/.docker 홈디렉토리 숨김 폴더 중 .docker 폴더 있는데 이 안에 config.json이 docker login 해도 반영 안되는 경우가 있어서 restart 때리거나 rm config.json 하고 다시 docker-login하면 문제 해결 rm ~/.docker/config.json docker login 하면 터미널에서 Docker-Hub 회원 인증 됨!!

원만한 서비스를 위해서 대부분의 작업들은 비동기로 수행 해야합니다 파이썬 기반 작업에서는 'Celery' 라는 녀석이 있습니다 채소 샐러리의 단면을 자르면 줄기와 심지가 많죠 파이프가 여러개라 여러 파이프라인으로 병렬(비동기)처리 한다고 생각하면 됩니다 오픈소스 개발자들의 기가 막히는 네이밍센스~ 그러면

의문:

멀티 쓰레딩보다 뭐가 낫냐 -> 연결 유실이나 실패 등 재실행 옵션을 줄 수있다. (max_retries)

구성요소 3가지

1. Broker : task를 worker에게 전달

2. Client : task 생성

3. Worker : task 수행

직관적이죠?

영화랑 드라마에서 브로커들이 항상 유통구조 중간에서 마약과 총기를 팔죠 중간 허브같은 지점입니다 개발에서는 업무를 보관하고 나눠주고!

Broker 종류?

1. RabbitMQ
2. Redis
3. Zookeeper(잘안씀) 일반적으로 RabbitMQ 와 Redis를 많이 씁니다 redis는 MQ, NoSQL, CacheDB 로 쓰이는 고가용성 플러그인입니다!! 다음 챕터엔 redis 실습 갈게요

airflow@0b77f1cca39a:/opt/airflow/logs/dag_id=binance/run_id=manual__2023-07-11T08:46:43.391781+00:00/task_id=binance_crawl$ sudo cat 'attempt=1.log'

We trust you have received the usual lecture from the local System Administrator. It usually boils down to these three things:

#1) Respect the privacy of others.
#2) Think before you type.
#3) With great power comes great responsibility.

[sudo] password for airflow: sudo: a password is required airflow@0b77f1cca39a:/opt/airflow/logs/dag_id=binance/run_id=manual__2023-07-11T08:46:43.391781+00:00/task_id=binance_crawl$ vi 'attempt=1.log' airflow@0b77f1cca39a:/opt/airflow/logs/dag_id=binance/run_id=manual__2023-07-11T08:46:43.391781+00:00/task_id=binance_crawl$ vi 'attempt=1.log' airflow@0b77f1cca39a:/opt/airflow/logs/dag_id=binance/run_id=manual__2023-07-11T10:18:13.187454+00:00/task_id=binance_crawl$ cat 'attempt=1.log' [2023-07-11T10:18:19.471+0000] {taskinstance.py:1103} INFO - Dependencies all met for dep_context=non-requeueable deps ti=<TaskInstance: binance.binance_crawl manual__2023-07-11T10:18:13.187454+00:00 [queued]> [2023-07-11T10:18:19.481+0000] {taskinstance.py:1103} INFO - Dependencies all met for dep_context=requeueable deps ti=<TaskInstance: binance.binance_crawl manual__2023-07-11T10:18:13.187454+00:00 [queued]> [2023-07-11T10:18:19.481+0000] {taskinstance.py:1308} INFO - Starting attempt 1 of 1 [2023-07-11T10:18:19.502+0000] {taskinstance.py:1327} INFO - Executing <Task(PythonOperator): binance_crawl> on 2023-07-11 10:18:13.187454+00:00 [2023-07-11T10:18:19.508+0000] {standard_task_runner.py:57} INFO - Started process 1315 to run task [2023-07-11T10:18:19.511+0000] {standard_task_runner.py:84} INFO - Running: ['airflow', 'tasks', 'run', 'binance', 'binance_crawl', 'manual__2023-07-11T10:18:13.187454+00:00', '--job-id', '6', '--raw', '--subdir', 'DAGS_FOLDER/binance_crawl.py', '--cfg-path', '/tmp/tmp7pzfq8yh'] [2023-07-11T10:18:19.513+0000] {standard_task_runner.py:85} INFO - Job 6: Subtask binance_crawl [2023-07-11T10:18:19.560+0000] {task_command.py:410} INFO - Running <TaskInstance: binance.binance_crawl manual__2023-07-11T10:18:13.187454+00:00 [running]> on host 0b77f1cca39a [2023-07-11T10:18:19.895+0000] {taskinstance.py:1547} INFO - Exporting env vars: AIRFLOW_CTX_DAG_OWNER='airflow' AIRFLOW_CTX_DAG_ID='binance' AIRFLOW_CTX_TASK_ID='binance_crawl' AIRFLOW_CTX_EXECUTION_DATE='2023-07-11T10:18:13.187454+00:00' AIRFLOW_CTX_TRY_NUMBER='1' AIRFLOW_CTX_DAG_RUN_ID='manual__2023-07-11T10:18:13.187454+00:00' [2023-07-11T10:18:20.231+0000] {taskinstance.py:1824} ERROR - Task failed with exception Traceback (most recent call last): File "/home/airflow/.local/lib/python3.7/site-packages/airflow/operators/python.py", line 181, in execute return_value = self.execute_callable() File "/home/airflow/.local/lib/python3.7/site-packages/airflow/operators/python.py", line 198, in execute_callable return self.python_callable(self.op_args, *self.op_kwargs) File "/opt/airflow/dags/binance_crawl.py", line 47, in _binance_api df.to_csv(f'/home/airflow/data/{file_name}', index=False) File "/home/airflow/.local/lib/python3.7/site-packages/pandas/core/generic.py", line 3482, in to_csv storage_options=storage_options, File "/home/airflow/.local/lib/python3.7/site-packages/pandas/io/formats/format.py", line 1105, in to_csv csv_formatter.save() File "/home/airflow/.local/lib/python3.7/site-packages/pandas/io/formats/csvs.py", line 243, in save storage_options=self.storage_options, File "/home/airflow/.local/lib/python3.7/site-packages/pandas/io/common.py", line 707, in get_handle newline="", PermissionError: [Errno 13] Permission denied: '/home/airflow/data/bccusdt_1month.csv' [2023-07-11T10:18:20.239+0000] {taskinstance.py:1350} INFO - Marking task as FAILED. dag_id=binance, task_id=binance_crawl, execution_date=20230711T101813, start_date=20230711T101819, end_date=20230711T101820 [2023-07-11T10:18:20.266+0000] {standard_task_runner.py:109} ERROR - Failed to execute job 6 for task binance_crawl ([Errno 13] Permission denied: '/home/airflow/data/bccusdt_1month.csv'; 1315) [2023-07-11T10:18:20.285+0000] {local_task_job_runner.py:225} INFO - Task exited with return code 1 [2023-07-11T10:18:20.303+0000] {taskinstance.py:2651} INFO - 0 downstream tasks scheduled from follow-on schedule check airflow@0b77f1cca39a:/opt/airflow/logs/dag_id=binance/run_id=manual__2023-07-11T10:18:13.187454+00:00/task_id=binance_crawl$ cat 'attempt=1.log' [2023-07-11T10:18:19.471+0000] {taskinstance.py:1103} INFO - Dependencies all met for dep_context=non-requeueable deps ti=<TaskInstance: binance.binance_crawl manual__2023-07-11T10:18:13.187454+00:00 [queued]> [2023-07-11T10:18:19.481+0000] {taskinstance.py:1103} INFO - Dependencies all met for dep_context=requeueable deps ti=<TaskInstance: binance.binance_crawl manual__2023-07-11T10:18:13.187454+00:00 [queued]> [2023-07-11T10:18:19.481+0000] {taskinstance.py:1308} INFO - Starting attempt 1 of 1 [2023-07-11T10:18:19.502+0000] {taskinstance.py:1327} INFO - Executing <Task(PythonOperator): binance_crawl> on 2023-07-11 10:18:13.187454+00:00 [2023-07-11T10:18:19.508+0000] {standard_task_runner.py:57} INFO - Started process 1315 to run task [2023-07-11T10:18:19.511+0000] {standard_task_runner.py:84} INFO - Running: ['airflow', 'tasks', 'run', 'binance', 'binance_crawl', 'manual__2023-07-11T10:18:13.187454+00:00', '--job-id', '6', '--raw', '--subdir', 'DAGS_FOLDER/binance_crawl.py', '--cfg-path', '/tmp/tmp7pzfq8yh'] [2023-07-11T10:18:19.513+0000] {standard_task_runner.py:85} INFO - Job 6: Subtask binance_crawl [2023-07-11T10:18:19.560+0000] {task_command.py:410} INFO - Running <TaskInstance: binance.binance_crawl manual__2023-07-11T10:18:13.187454+00:00 [running]> on host 0b77f1cca39a [2023-07-11T10:18:19.895+0000] {taskinstance.py:1547} INFO - Exporting env vars: AIRFLOW_CTX_DAG_OWNER='airflow' AIRFLOW_CTX_DAG_ID='binance' AIRFLOW_CTX_TASK_ID='binance_crawl' AIRFLOW_CTX_EXECUTION_DATE='2023-07-11T10:18:13.187454+00:00' AIRFLOW_CTX_TRY_NUMBER='1' AIRFLOW_CTX_DAG_RUN_ID='manual__2023-07-11T10:18:13.187454+00:00' [2023-07-11T10:18:20.231+0000] {taskinstance.py:1824} ERROR - Task failed with exception Traceback (most recent call last): File "/home/airflow/.local/lib/python3.7/site-packages/airflow/operators/python.py", line 181, in execute return_value = self.execute_callable() File "/home/airflow/.local/lib/python3.7/site-packages/airflow/operators/python.py", line 198, in execute_callable return self.python_callable(self.op_args, *self.op_kwargs) File "/opt/airflow/dags/binance_crawl.py", line 47, in _binance_api df.to_csv(f'/home/airflow/data/{file_name}', index=False) File "/home/airflow/.local/lib/python3.7/site-packages/pandas/core/generic.py", line 3482, in to_csv storage_options=storage_options, File "/home/airflow/.local/lib/python3.7/site-packages/pandas/io/formats/format.py", line 1105, in to_csv csv_formatter.save() File "/home/airflow/.local/lib/python3.7/site-packages/pandas/io/formats/csvs.py", line 243, in save storage_options=self.storage_options, File "/home/airflow/.local/lib/python3.7/site-packages/pandas/io/common.py", line 707, in get_handle newline="", PermissionError: [Errno 13] Permission denied: '/home/airflow/data/bccusdt_1month.csv' [2023-07-11T10:18:20.239+0000] {taskinstance.py:1350} INFO - Marking task as FAILED. dag_id=binance, task_id=binance_crawl, execution_date=20230711T101813, start_date=20230711T101819, end_date=20230711T101820 [2023-07-11T10:18:20.266+0000] {standard_task_runner.py:109} ERROR - Failed to execute job 6 for task binance_crawl ([Errno 13] Permission denied: '/home/airflow/data/bccusdt_1month.csv'; 1315) [2023-07-11T10:18:20.285+0000] {local_task_job_runner.py:225} INFO - Task exited with return code 1 [2023-07-11T10:18:20.303+0000] {taskinstance.py:2651} INFO - 0 downstream tasks scheduled from follow-on schedule check airflow@0b77f1cca39a:/opt/airflow/logs/dag_id=binance/run_id=manual__2023-07-11T10:18:13.187454+00:00/task_id=binance_crawl$

Docker-compose로 환경 구축하면 서버 금방 띄울 수 있음!

DAG개발 외에 관심사를 줄이기 위해 개발 환경을 Docker-compose로 구축 공식홈페이지의 yml 파일 참조

Airflow 서버를 띄우는 시간 단축 : 5분 -> 1분

개발 피드백 루프 시간 단축 : 1분(Commit -> Sync) -> 5초

개발 클러스터에서 유휴 Airflow들이 Node를 점유하였던 문제 해결 : 2개 이상의 VM 절약

Docker라는 표준 환경을 통해 Airflow 서버의 불안정성을 낮추고 관리 비용을 줄임

웹서버 / 스케쥴러 / DB

웹서버(Web Server):

Airflow 웹서버는 사용자 인터페이스(UI)를 제공하고 데이터 파이프라인의 모니터링, 스케줄링, 실행 등의 작업을 관리합니다. 웹서버는 사용자가 DAG(Directed Acyclic Graph)를 정의하고 관리할 수 있도록 도와줍니다. 웹 인터페이스를 통해 작업의 실행 상태, 로그, 스케줄러 등을 확인하고 관리할 수 있습니다.

스케줄러(Scheduler):

Airflow 스케줄러는 정의된 DAG를 기반으로 작업 실행을 스케줄링합니다. 스케줄러는 설정된 실행 주기 또는 종속성을 기준으로 DAG의 작업들을 적절한 시간에 실행합니다. 스케줄러는 작업 상태를 추적하고 작업 간의 의존성을 관리하여 데이터 파이프라인의 실행을 조정합니다.

Executor:

Executor는 Airflow에서 실행 중인 작업들을 처리하는 방식을 정의합니다. Airflow는 LocalExecutor, CeleryExecutor, KubernetesExecutor 등 다양한 Executor를 지원합니다. 각 Executor는 작업을 병렬로 실행하고 관리하는 방식이 다르며, 작업 처리량과 성능에 영향을 줄 수 있습니다.

메타데이터 데이터베이스(Metadata Database):

Airflow는 작업, DAG, 실행 상태 등의 메타데이터를 저장하기 위한 데이터베이스가 필요합니다. 메타데이터 데이터베이스는 작업 스케줄링, 실행 상태 추적, 작업 히스토리 등을 관리하는 데 사용됩니다. 일반적으로 SQLite, PostgreSQL, MySQL 등의 데이터베이스가 사용됩니다.

데이터레이크

AWS의 서비스들로 구성해놨습니다.

1. Cloud Storage ( AWS S3, GCS, Azure Blob Storage)

2. ETL ( AWS Glue)

3. 처리 및 분석( AWS EMR : Elastic MapReduce)

4. 카탈로그 및 메타데이터 관리 ( AWS Glue Data Catalog)

위의 클라우드 서비스들을 보면 알겠지만, 데이터레이크는 비정형(raw)데이터 를 모아놓는 것이다!!!

정적 파일(jpg) 로그 데이터, 센서 데이터 등등

근데 웹서버(nginx) conf 파일 Location에 staticfile(html) 등을 모아 놓는 것은

데이터 레이크가 아니다~ 아시죠 ?!?!?

데이터 웨어하우스

전사 차원에서 BI(Business Intelligence)로 활용하기 위해

데이터 마트

마트는 웨어하우스보다 작죠? 그래서 좀더 작은 단위인 부서에서 자체 구축하는 사례입니다

위의 DW / DM 구축을 클라우드 플랫폼(paas) 에서 구축한다면..?

돈이 많이 들지만, 책임 전가가 가능하고, 안정성이 높고 확장성이 용이하겠죠 ?ㅎㅎ

그래서 무슨 서비스냐?

AWS Redshift,

Google BigQuery,

Azure Synapse Analytics

가 있습니다.

클라우드 서비스를 빗대서 요약하면 데이터 레이크는 raw 데이터가 S3를 씁니다. 용도에 따라 맵리듀스 붙일 수도 있는데, 자체 하둡 구축할 수도 있고, EMR 쓸 수도 있습니다. 웨어하우스는 BI합니다. 마트는 부서에서 분석용으로 씁니다. RedShift 쓰면 서버리스 서비스라 저렴합니다 !!(비용 산정은 따로 해봐야합니다.)

S3, Redshift, EMR 짱이다!!

m1 , pytorch GPU 활성화하기

m1, m2 : arm64 아키텍처 사용자라면 텐서플로우와 파이토치로 로컬에서 GPU를 사용하고 싶고 사용할 수 있다!! 하지만 어떻게~?!?! tensorflow에 관한 게시글은 너무 많아서 패스하고 토치에 대해 간단하게 설명을 하면 일단 Tensorflow 같은 경우는 conda설치( Miniforge) 후에 가상환경 만들고 관리하고 환경변수 설정도 해줬던 것 같음 ( bashrc, export ) 등등 pytorch도 비슷하게 conda에 가상환경 만들었는데 안되는거임 버전이슈인가 했는데. 그래서 그냥 conda 가상환경에서 나오고

깨알 conda 명령어
가상환경 종류 확인
conda info --envs
가상환경 생성
conda create --name <이름> <옵션줄수있음,파이썬버전등,나는잘안줌)
가상환경 나오기
conda deactivate

로컬에 가상환경 python -m venv <가상환경이름> 해서 거기서

pip install torch
python

쉘 드가자!!

import torch
torch.backends.mps.is_avaiable()

True 뜨면 끝! GPU 활성상태 확인은 is_available() 구절을 참 좋아함 ㅋ

flume 기본

여러 was, pod라고 볼수도 있고, 각 was에 Flume Agent를 배치하고 flume으로 갖고오면 Source -> Channel -> Sink 과정을 거쳐서

1. 또 다른 Flume Agent를 줄 수도 있고,
2. ELK(Elastic-Search, Logstash, Kibana) 줄 수도 있고,
3. 하둡+(하이브,슈마허) 줄 수도 있음. 즉 Message Broker 느낌도 있는듯?

flume이 kafka랑 비슷해보이는데... 그럼 차이는?

Flume:

데이터 수집 및 로그 전송을 위한 분산 시스템입니다. Flume은 로그 데이터를 수집하고 신뢰성 있는 방식으로 중앙 집중형 데이터 스토리지 또는 분산 시스템으로 전송합니다. Agent, Source, Channel, Sink 등으로 구성되며, Source는 데이터 소스에서 데이터를 수집하고, Channel은 데이터를 보관하며, Sink는 데이터를 대상 시스템으로 전송합니다. 주로 로그 파일, 디렉토리, 웹서버 로그 등의 데이터를 수집하고, Hadoop HDFS, Apache HBase, Elasticsearch 등과 같은 시스템으로 전송할 때 사용됩니다.

Kafka:

분산 스트리밍 플랫폼으로, 대량의 데이터를 실시간으로 처리하는 데 중점을 둡니다. 메시지 큐 시스템으로, 데이터 스트리밍과 이벤트 처리를 지원합니다. 데이터는 Topic으로 구성된다. Producer가 데이터를 Topic에 전송하고, Consumer가 Topic에서 데이터를 읽을 수 있습니다. 데이터는 영속적으로 보관되며, 다양한 Consumer가 동일한 데이터를 읽을 수 있습니다. 스케일 아웃 및 고가용성에 강점을 가지며, 실시간 데이터 처리, 로그 분석, 웹 애플리케이션 모니터링 등에 사용됩니다. 요약하자면, Flume은 데이터 수집과 전송에 중점을 둔 분산 시스템이고, Kafka는 대량의 실시간 데이터 처리와 이벤트 스트리밍을 위한 분산 스트리밍 플랫폼입니다. Flume은 데이터를 수집하여 다른 시스템으로 전송하는 데 사용되며, Kafka는 메시지 큐 시스템으로 데이터를 실시간으로 처리하고 스트림으로 전달하는 데 사용됩니다. 선택은 사용 사례, 요구 사항 및 아키텍처에 따라 달라집니다.

하둡

파일 생성, 삭제, append 만 가능 (즉, 수정 불가)

NO MODIFY!!

구입한 도메인이 '네임서버' -> .com .net .co.kr .org

3. Route 53 서버 연결하기 (S3) 2개 네임서버 (www 붙이고 / 안 붙이고) www.웹사이트이름.com 웹사이트.com S3: 정적 웹사이트 호스팅(nginx 에서 location에 담아 놓은 static 파일: html템플릿 서빙)
   1. Overview에 html 파일들 담기
   2. Properties 에 Static website hosting 활성화
   3. Index document에 업로드한 파일 이름 입력(index.html)
   4. Permissions 탭 Bucket Policy에서 GetObject 허용하는 Policy JSON 작성

Linux

파일 만들고 쓰기
vi <filename>
-파일 만들고만 싶을 때
touch <filename>
touch와 vi의 차이는 vi는 파일을 만들고 쓰러 들어가고
touch는 만들어만 놓는다. 쓰러 안들어감( 톡, 터치해서 만들기만한다고 생각하면 편함)
-디렉토리(폴더)를 만들고 파일을 만들때 종종쓰는 커맨드
test폴더에 test1, test2, test3 파일을 만들기
test폴더도 만들어야 한다
-폴더 만들기
mkdir test
-파일 만들기
cd test
touch test1.py
touch test2.py
touch test3.py
간략하게 써보면
mkdir test && cd test && touch test1.py test2.py test3.py && cd ..
1.폴더생성 - > 2. 폴더로 이동 -> 
3. 파일 3개 만들기(띄어쓰기하면 여러개 생성할 수 있음)-> 4. 다시 돌아오기
- 파일 지울때
remove 커맨드
rm -r (--recursive 재귀적으로 삭제)
rm -f 강력한 삭제
rm -rf 완전 강력한 삭제
폴더, 파일 다 삭제 가능
정리를 하면
파일 하나 없앨 때는
rm <filename>
폴더를 지울 때는(파일만 삭제도 가능함)
rm -r <foldername>
폴더 안에 있는 파일이 여러개여서 이걸 재귀적으로 방문하면서 지우는 거라
-r옵션임
근데 잘안지워지면
rm -rf <foldername>
현재 위치에서 여기에 있는 모든 파일, 폴더를 지우고 싶을때
rm -rf *
*은 항상 와일드 카드 느낌
rm -i <filename>
i 옵션은 파일을 삭제할때 os가 저에게 재차 확인합니다
삭제하시겠습니까? 대화를 하기때문에 i는 interactive의 약자입니다.
rm -rif
rm -irf
rm -ifr 
다 똑같습니다 -에 옵션주는 건 순서 상관 없이 들어갑니다
즉 위의 명령어는 "강력하게" "재귀적으로" "삭제할지 물어보며"
삭제하는 명령어입니다.
-뒤에 옵션 여러개 주는 것과 순서가 상관 없단 것 중요!