Taegu

airflow설치 후 다양한 테스트 도중 아래와 같은 문구가 생겼습니다.

scheduler does not appear to be running 문구를 본다면 스케쥴러 데몬이 내려간 것처럼 보이지만 airflow 스케쥴을 정상적으로 작동하고 있는 상태였습니다.

이러한 문구가 보이는 이유는  Sequential Executor를 사용하기 때문입니다. 

 Sequential Executor란 기본 Executor로 sqlite와 함께 사용할 수 있으며, 한번에 하나의 task만 실행할 수 있어 병렬성을 가지지 않습니다.

따라서, 현재 task를 실행중인 상태에서는 heartbeat를 받을 수 없어 생성되는 문구라고 생각하면 됩니다.

최근 MSSQL에 python을 통해 전처리 한 대용량 데이터를 insert할 일이 생겨, 관련된 내용을 정리하려고 합니다.

테스트에 사용하는 데이터의 row는 10만건이며 행은 15행이 존재합니다.

TEST1 : 6분26초 (하나씩 insert - 최종 commit)

첫 번째 테스트는 데이터를 하나씩 삽입 후에 commit을 진행하였습니다.

10만건의 데이터를 삽입하는데 10분이 걸렸으며, 모든데이터 삽입 후 commit을 하기 때문에 중간에 에러 발생시 롤백이 진행됩니다.

TEST2 : 18분41초 (하나씩 insert - 삽입 마다 commit)

 두 번째 테스트는 TEST1과 같은 방식이지만 하나씩 삽입할 때마다 커밋을 해주었습니다. 이런경우 중간에 에러가 날 경우라도 기존에 삽입된 데이터는 유지된다는 장점이 있지만 TEST1에 비해 약 3배 정도 느려진 것을 확인할 수 있습니다.

TEST3 : 23초 (bulk insert)

 세 번째 테스트는 한번 삽이할 때 30개의 데이터를 삽입하는 방법입니다. 벌크 인서트라고 하며, 기존 TEST 대비 엄청 빠른 속도를 보입니다. 

저는 아래 답변을 참고하였습니다.

https://hashcode.co.kr/questions/10863/python-%EC%9A%A9%EB%9F%89%EC%9D%B4-%ED%81%B0-dataframe%EC%9D%84-mssql%EC%97%90-%EB%B9%A0%EB%A5%B4%EA%B2%8C-insert%ED%95%A0-%EC%88%98-%EC%9E%88%EB%8A%94-%EB%B0%A9%EB%B2%95%EC%9D%B4-%EB%AD%94%EA%B0%80%EC%9A%94

TEST4 : 7분 38초 (executemany)

executemany 말 그대로 많은 데이터를 인서트하는 방법입니다. 하지만 제가 잘 이해를 못해서 그런지 성능이 크게 좋지 못했습니다.

결론

앞선 테스트에서 봤듯이 저는 TEST3번을 통해 데이터를 INSERT했습니다.

TEST3 방법을통해 약 190만건의 데이터를 INSERT하는데 7분41초가 걸렸으며, 혹시 이보다 빠른 속도를 원하시는 분들은 sqlalchemy(create_engine), to_sql등 찾아보면 좋을 것 같습니다.

airflow 설치 후 example dag를 삭제하고 싶으신 분들이 있을겁니다. 이번 포스팅은 example dag 삭제해보겠습니다.

airflow 스크립트 수정

conda activate test
cd airflow
#load_examples = True를 Fasle로 변경
vi airflow.cfg

vi로 airflow.cfg를 열어 load_examples=True를 load_examples=False로 변경해줍니다.

airflow db reset 및 webserver 재실행

# airflow 폴더에서 나옵니다.
cd

# airflow 프로세스가 살아있는지 확인 살아있을 경우 kill -9 프로세스아이디 를 통해 종료
ps -ef |grep 'air'
 
# db초기화
airflow db init

# webserver 및 scheduler 재실행
nohup airflow webserver -p 8080 &
nohup airflow scheduler &

airflow 확인

localhost:8080 접속하게 되면 example dag가 없어진걸 볼 수 있습니다.

아나콘다 설치

아나콘다 설치 후 airflow를 설치해보겠습니다. 아나콘다 설치가 안된분들은 아래 포스팅을 확인해주새요

https://taeguu.tistory.com/100

airflow 설치

# conda 가상환경 실행
# 가상환경이 없는경우 conda create -n test pip python=3.7 명령어 실행
conda activate test

# airflow 설치
pip install apache-airflow

# db init
airflow db init

# webserver 실행
airflow webserver -p 8080

airflow 웹서버 접속

웹서버를 실행하게 되면 아래와 같이 airflow 웹서버거 실행되며 localhost:8080을 url에 입력하여 접속합니다.

접속하면 아래 그림과 같이 로그인 창이 뜹니다. 이제 다시 커멘드창으로 이동하여 계정을 생성하겠습니다.

airflow 계정 생성

웹서버를 창은 그대로 두고 새로운 창을 열어 아래 명령어을 입력하게 되면 admin권한을 가지는 admin계정이 생성됩니다.

airflow users create  --username admin  --firstname admin  --lastname admin   --role Admin  --password admin  --email admin

airflow 계정접속

id / pw : admin / admin으로 접속하게 되면 sample dag가 보이며, 접속이 완료됐습니다.

airflow 둘러보기

특정 dag를 클릭하게 되면 dag안에 task들의 의존성 및 dag 코드를 확인할 수 있습니다.

이번 포스팅은 리눅스 서버에 아나콘다를 설치하겠습니다.

os는 우분투 18.04 lts에 설치해보도록 하겠습니다.

아나콘다 다운

https://www.anaconda.com/products/distribution#linux

우선 위의 링크에 접속해 anaconda 리눅스 버전을 다운받겠습니다.

마우스 우 클릭 후 링크 주소를 복사합니다.

curl -O https://repo.anaconda.com/archive/Anaconda3-2022.10-Linux-x86_64.sh

위 명령어를 통해 아나콘다를 다운로드 합니다.

아나콘다 설치

1. sh파일 실행

bash Anaconda3-2022.10-Linux-x86_64.sh

2. 실행 후 엔터를 누르다 보면 라이센트 관련 내용이 나오고 yes를 입력합니다.

3. 아나콘다 디렉토리 위치 선택이 나오며, 엔터를 누르면 자동으로 설정된 디렉토리에 아나콘다가 설치됩니다.

4. conda init 실행 질문에 yes

아나콘다 확인

 설치가 완료된 후 conda 명령어가 실행되지 않을 수 있다 그럴 경우 아래 명령어를 입력해 줍니다.

home아래 본인의 디렉토리 입력 필수

export PATH="/home/taegwan/anaconda3/bin:$PATH"

입력 후에 cond 작성후 tab버튼을 눌렀을 때 자동완성으로 conda가 입력되면 아나콘다가 설치된 것입니다.

또한 서버 재접속 시 명령어 앞에 (base)가 표시되면 설치가 정상적으로 된 것입니다.

마지막으로 아래 명령어를 통해 콘다 가상환경을 생성한 후 pip 버전을 확인해보겠습니다.

conda env list
#create 실행 후 y입력
conda create -n test python=3.8
conda env list
conda activate test
pip --version

다음은 jupyter notebook 설치를 통해 원격으로 접속하는 방법을 포스팅하겠습니다.

위 에러는 문장에 특정 단어를 strip으로 제거할 때 나타났습니다. 문장 그대로 보면 str이 아닌 bytes-like object가 필요하다고 하지만, 에러가 난 문장은 byte타입이었으며, 문자열로 변경하니 정상적으로 작동하는 것을 확인할 수 있었습니다.

에러

해결

현재 국내 상장중인 코스피 코스닥 기업을 알아보도록 하겠습니다.

라이브러리

#주식정보 라이브러리
import FinanceDataReader as fdr

#시각화 라이브러리
import matplotlib.pyplot as plt
import seaborn as sns
plt.rc('font', family='NanumGothic')

#전처리 라이브러리ㅣ
from tqdm import tqdm_notebook
import pandas as pd

코스닥 종목 불러오기

FinanceDataReader 라이브러리를 통해 코스닥 종목을 불러왔습니다. 상장일이 NA값인 종목을 제거하니 1,577건의 코스닥 상장종목을 확인할 수 있으며, 주식명, 섹터, 산업, 대표이름, 홈페이지 등의 정보를 확인할 수 있습니다.

#코스닥 종목 불러오기
df_kosdaq = fdr.StockListing('KOSDAQ')
#결측값 제거
df_kosdaq = df_kosdaq[df_kosdaq['ListingDate'].notna()]
print(df_kosdaq.shape)
df_kosdaq.head()

코스닥 상위 top10 섹터

코스닥에 상장된 섹터 중 상위 10개의 섹터를 불러온 후 시각화하였습니다. 소프트웨어 개발 및 공급업과 특수 목적용 기계 제조업 섹터의 상장이 많은 것을 볼 수 있습니다.

li = df_kosdaq['Sector'].value_counts()[:10].index
top10 = df_kosdaq[df_kosdaq['Sector'].map(lambda x : x in li)]
plt.figure(figsize = (20,10))
sns.countplot(data = top10,y= 'Sector')
plt.show()

코스피 종목 불러오기

코스피 종목을 불러오겠습니다. 위와 다르게 'KOSPI'를 입력하게 되면 코스피 종목들을 불러올 수 있습니다. 코스피 종목들은 822건으로 확인됩니다.

#코스피 종목 불러오기
df_kospi = fdr.StockListing('KOSPI')
#결측값 제거
df_kospi = df_kospi[df_kospi['ListingDate'].notna()]
print(df_kospi.shape)
df_kospi.head()

코스피 상위 top10 섹터

코스피의 경우 코스닥과 달리 기타 금융업과 자동차 신품 부품 제조업이 많은 것으로 나타났습니다.

li = df_kospi['Sector'].value_counts()[:10].index
top10 = df_kospi[df_kospi['Sector'].map(lambda x : x in li)]
plt.figure(figsize = (20,10))
sns.countplot(data = top10,y= 'Sector')
plt.show()

도커  컨테이너 내부 접속

아래와 같이 명령어를 입력하면 도커 내부 컨테이너에 접속할 수 있습니다.

docker exec -it [컨테이너 ID] /bin/bash