imhamburger 님의 블로그

에어플로우에는 다양한 오퍼레이터를 제공한다. 이번엔 지난글에서 다룬 BashOperator말고 PythonOperator를 다뤄보자. 1. PythonOperatorfrom airflow.operators.python import PythonOperatordef print_context(ds=None, **kwargs): print("::group::All kwargs") task = PythonOperator( task_id="print_the_context", python_callable=print_context )PythonOperator는 Dags에 Python 함수(내가 만든 기능)를 입력하고 task 에서 py..

API를 통해 어떤 데이터를 불러올 때 나에게 발급된 키를 공개하지않기 위해 os.getenv를 사용한다.깃헙에 나의 코드를 push할 때 나의 개인 키가 공개되어서는 안되기 때문이다. 따라서 환경변수를 불러올 수 있는 os.getenv를 사용할 수 있다.사용하는 방법은 우선 로컬환경에서 키를 지정한다. 예를들어 "abcdef"라는 키를 발급받았고 해당 키를 이용해야한다고 하였을 때,터미널창에 아래와 같이 입력한다.export API_KEY="abcdef"API_KEY라는 변수에 내 키값을 담는다. 그럼 파이썬파일에서 다음과 같이 이용할 수 있다.import osdef get_key(): key = os.getenv('API_KEY')os.getenv는 로컬 환경에서 변수를 불러오기 때문에 내가 지..

pdm은 이전글에서도 설명한대로 의존성 관리도구이며 pyenv처럼 가상환경 기능을 제공한다.여러가지 모듈을 pdm 가상환경에 설치하여 다양한 기능을 테스트해 볼 수 있다. 당연히 의존성 관리 도구이기 때문에 설치한 모듈을 관리할 수 있다. 관리할 수 있다는건 생성, 삭제, 수정이 용이하다는 것이다. pdm 가상환경을 구축하는 건 매우 간단하다. pdm을 실행하고자 하는 경로로 이동하여 아래 코드를 차례대로 입력하면 된다.pdm init #pdm 실행source .venv/bin/activate #pdm 가상환경 실행 만약 가상환경이 실행이 되지 않는다면 venv가 없기때문이다. 아래 코드로 따라서 별도로 설치하고 다시 실행하면 된다.pdm venv create #venv 생성 여기 드는 의문! 그냥 가상..

파라미터값에 default를 둘 수 있으며, 보통 많이 사용하는 파라미터값을 default로 설정한다.def gen_path(filename, path='/data/csv', home='~/my/airflow'): print(f"{home}{path}/{filename}") >>> gen_path('file') #입력값>> ~/my/airflow/data/csv/file #출력값위 코드에서 'path'와 'home'을 default값으로 각각 경로를 설정하였다.그럼 입력할 때 이미 default 값이 설정되어 있기때문에 출력값은 입력값과 default값이 합쳐져 나온다. 만약 default값이 앞으로 온다면?에러가 날 것이다. default 파라미터값은 앞에 올 수 없다.def add..

지난주에 에어플로우 Operator에서 csv파일을 만드는 것까지 진행하였다. (지난글 바로가기)이번주엔 csv파일로 만든 것을 sql을 이용해 데이터베이스에 저장하고 판다스를 이용해 parquet 파일형태로 다시 저장해보는 것을 하였다. 약간 과정이 복잡하기는한데... 차근차근 알아보자. 4. task_create_tbl(table)task_create_tbl = BashOperator( task_id="create.table", bash_command=""" SQL={{var.value.SQL_PATH}}/create_db_table.sql MYSQL_PWD='{{var.val..

partition_cols 을 이용하면 데이터 용량을 줄일 수 있다.partition_cols은 일반적으로 데이터프레임을 저장할 때 특정 열(Column)을 기준으로 데이터를 분할하는 데 사용되는 인수이다. partition_cols은 보통 Parquet 또는 ORC 파일 형식 저장에서 사용되며, 데이터를 여러 파일 또는 폴더로 나눠서 저장한다.이렇게하면, 방대한 데이터가 있을 때 더 관리하기 쉬워지고 데이터를 빠르게 찾아 로드할 수 있다. 게다가 큰 장점은 용량이 적다는 것! 아래는 실제로 partition_cols을 사용하여 저장한 데이터와 그렇지 않은 데이터의 용량 크기 차이를 보여준다. Partition_cols을 사용하는 방법은 간단하다.{데이터테이블명}.to_parquet('{저장할 파일경로}..

내가 만들고자 했던 것은,dtcmdcnt날짜명령어개수3개의 컬럼으로 이루어진 데이터 테이블을 판다스로 parquet 파일을 만들어 저장하고 터미널에서 cmd 를 입력하면 cmd에 해당하는 cnt 즉 개수가 출력되는 기능을 만들고자 했다. 여기서 cmd를 인자로 받아 cnt가 출력되게 만들어야 한다. 따라서 sys를 이용해 다음과 같은 파이썬 코드를 작성하였다.import pandas as pdimport sysa = sys.argv[1]def cnt(): df = pd.read_parquet("~/tmp/history.parquet") fdf = df[df['cmd'].str.contains(a)] cnt = fdf['cnt'].sum() print(cnt)인자값을 0이 아닌 1을 ..

판다스(Pandas)를 사용하기 위해 주피터노트북을 사용하였다. pyenv로 가상환경을 만들고 그 안에 주피터노트북을 설치하였다. 1. 주피터노트북을 실행하고 pandas를 import한다.import pandas as pd 2. csv파일로 된 데이터를 불러오기 위해 다음과 같이 입력한다.df = pd.read_csv('{파일경로}', on_bad_lines='skip')on_bad_lines='skip'을 추가한 이유는 내가 가지고있는 csv파일 안에 형식이 안맞는 데이터들이 좀 있어서 스킵하고 사용하였다. 인코딩 에러메세지가 떴다. 해결방법은 아래 코드처럼 encoding = "latin"을 넣어주면 된다.df = pd.read_csv('~/data/csv/240717/csv.csv', on_bad..

시작하기 전에.. 터미널에 아래와 같이 입력하면 비밀번호 입력없이 Mysql에 접속할 수 있다.MYSQL_PWD='{비밀번호 입력}' mysql -u root   mysql로 csv파일로 불러올 때, 여러 에러메세지들이 발생하였다. 차근차근 해결해보자! 에러메세지 1 - 백슬러시ERROR 1261 (01000) at line 1: Row 13 doesn't contain data for all columns 해당 에러는 백슬러시 \ 때문에 발생하는 문제로 데이터에 \가 포함되어 있기 때문이다. 이를 무시하게끔 만드려면 FIELDS TERMINATED BY 옆에 혹은 아래에 ESCAPED BY '' 코드를 추가하면 된다.ESCAPED BY ''   에러메세지 2 - 인코딩해당 에러는 인코딩문제로 나타나는 ..

맥 터미널에서 아래와 같은 코드를 사용했는데, cut: stdin: Illegal byte sequence 에러가 났었다.cat ~/data/test.txt | cut -d' ' -f1 이 에러는 터미널에서 UTF-8이 아닌 다른 종류의 인코딩된 데이터를 처리하려고 할 때 발생한다. *UTF-8(Unicode Transformation Format - 8-bit)은 유니코드 문자 인코딩의 한 방식으로 UTF-8은 전 세계 모든 문자를 표현할 수 있도록 설계된 가변 길이 문자 인코딩 방식이다. 이를 해결하기 위해서는 터미널 창에 아래 코드를 입력하면 된다.export LC_ALL=C 매번 같은 에러가 난다면, 영구적으로 설정할 수도 있다. ~/.zshrc 에 위에 코드를 추가하면 된다. 그럼 매번 터미널에..