안녕하세요. 주형권입니다. 최근에 개발을 다하고 시간이 조금 남아서 개발하면서 몇 가지 만났던 부분에 대해서 글을 많이 쓰게 되었습니다. 지금은 ELT 파이프라인을 개발하고 있는데요. 저 같은 경우 주로 Airflow를 통해서 스케줄 처리를 하고, PythonOperator를 이용해서 Python Class를 불러와서 데이터를 처리하는 구조로 ELT 파이프라인을 만듭니다. 이렇게 만들면 제가 입맛데로 원하는 가공을 할 수도 있고 여러 가지 기능을 제가 원하는 방향으로 넣을 수 있어서 PythonOperator를 선호하고 있습니다. 이 ELT 파이프라인은 조만간 정리하여 글을 쓰도록 하겠습니다. (거의 완성되었습니다.) 무엇을 하려고 하였는가? Airflow에 Connection에 Oracle 서버를 등록..

안녕하세요. 해당 글을 MySQL 문법을 사용 하였습니다. 지난번에 Query를 통해서 Airflow의 스케줄 시간을 파싱 하는 글을 작성하였는데요. ( 바로가기 ) 이번에는 파싱하여 일반 사용자가 보기 편하게 나타내는 Query를 만들었습니다. 위와 같이 Crontab시간을 사용자가 보기 편한 형태로 보여 주도록 하였습니다. 아무래도 규치적을 주고 파싱을 하다 보니 영어 표현과 어순이 조금 어색하거나 안 맞는 경우가 있습니다. 이 부분은 감안해 주세요. (도저히.. 어떻게 바꿀지 모르겠습니다.) SELECT schedule_interval ,CASE WHEN schedule_interval = '"@once"' THEN 'Schedule once and only once' WHEN schedule_in..

안녕하세요. 데이터엔지니어 주형권입니다. 오랜만에 꽤나 길고 범용적인 주제에 관해서 글을 쓰려고 합니다. 많은 회사에서 데이터를 활용하여 많은 업무를 하고 데이터를 이용해서 많은 의사결정을 하고 있습니다. 그러면서 데이터의 양도 방대하고 종류도 꽤나 다양해졌습니다. 제가 처음에 일할 때는 RDB(Relational Database)에서 발생하는 데이터만 주로 다루었습니다. 하지만 최근에는 너무도 많고 너무도 다양한 데이터를 이용해서 의사결정과 업무를 진행합니다. 하지만 데이터가 아무리 다양하고 많아져도 RDB에서 발생하는 트랜잭션 데이터는 절대적으로 필요합니다. 그렇기에 RDB에서 데이터를 가져오는 것은 절대로 필요하다고 볼 수 있습니다. 회원 데이터 또는 빌링 데이터등은 무결성이 무조건이기 때문에 RD..

글을 읽기 전에 이 내용은 저의 경험을 토대로 작성하였습니다. 현재 사용하시는 도구 및 방법과 다르다고 하여 무엇이 맞고 틀리고를 이야기하고자 하는 글이 아닌 경험을 공유하고자 작성된 글이므로 참고 차원에서 봐주시길 바랍니다. 데이터를 보기 위해서는 많은 과정이 필요합니다. SQL을 이용해서 데이터를 추출하고 이를 엑셀로 다운로드하거나 리포트를 만들어서 보는 과정이 아닌 전체적인 과정을 설명하고자 합니다. 이 글의 내용은 제가 아는 범위에서 작성하였으나 회사마다 모두 다르고 사람마다 모두 다른 방식으로 할 수 있기 때문에 어느 정도 감안해서 글을 읽어 주시길 바랍니다. 보통 데이터는 크게 3가지의 데이터가 있습니다. 정형 데이터 : 관계형 데이터베이스 (MySQL , SQL Server , Oracle)..

안녕하세요. 이 내용을 쓸지 말지 정말 고민을 많이 했습니다. 예전에 링크드인에 관련하여 포트폴리오 형식으로 PPT를 만들어서 올렸는데요. 조회수가 3만에 육박하여 굉장히 많은 분들이 관심을 가져 주셨습니다. 그래서 매번 써야지 써야지 했는데 내용이 워낙 길어서 고민을 많이 했습니다. 간단하게 쓸 수 없었습니다... 1.0 개발 - 2개월 2.0 개발 - 2개월 3.0 개발 - 2개월 이렇게 총 3번의 버전 업데이트를 하였습니다. 처음에는 데이터를 볼 수 없어서 분석을 하지 못하였는데, BigQuery로 넣는 작업부터 시작하여... 재사용성을 고려하여 코드를 리팩토링 하였고, 마지막으로 multi thread를 적용하였습니다. 코딩을 2019년에 처음 시작하여, 지금의 회사에서 팀장님과 팀원들에게 많이 ..

💁‍♂️ 들어가며 데이터를 적재하면서 가장 중요한 것은 무엇일까요? 여러가지 이유가 있겠지만 무엇보다 그 데이터가 정상적으로 잘 적재되었는지 여부 입니다. 많은 사람들이 적재를 어떻게 하면 "빠르고 효율적으로 적재 할 수 있나?" 에 초점을 많이 둡니다. 하지만 이 데이터가 정상적으로 A 👉🏻 B 들어왔는지는 많이 고민하지 않는 것 같습니다. 이 글에서는 실제로 구현한 코드를 보여주진 않습니다. (회사에서는 제가 구현을 했지만..) 컨셉을 주로 설명하며, 각자가 맞게 회사 시스템에 녹일 수 있으면 좋겠습니다. 🤷‍♂️ 어떻게 체크 할것인가? 데이터를 가공하여 계산 된 데이터를 적재하는 경우는 그 데이터가 정확하게 나온 지 여부를 확인하지만 원천 데이터에서 데이터를 가져올 경우 데이터를 맞게 가져왔는지 확..

기존에 링크드인에 공유한 포토폴리오 형식으로 제작 된 데이터 파이프라인 제작 관련 자료 입니다. 관련하여, 다운로드를 받으시려면 가장 아래에 첨부 파일을 확인해 주세요. 데이터 파이프라인 PDF

오랜만에 글을 쓰는거 같습니다. 한동안 서버 세팅을 하면서 바쁜 시간을 보내고 재택 근무를 하면서 이런저런 적응을 하다보니 글을 안쓰게 되었습니다. 그러다 이번에 새롭게 Airflow2.0을 실제로 세팅 하면서 이런 저런것을 정리해야겠다 싶어서 글을 씁니다. 들어가며... 이 글을 읽고 참고 하시면 도움이 될 분들은 이런 분들 입니다. 스케줄 처리를 해야하는데, Airflow를 선택 하였다. 하지만 나는 딥하게 알지는 못한다. 대규모 처리가 필요한게 아닌 적당한 처리를 할 예정이다. 일단 실무에 급하게 반영해야 하는데 봐도 잘 모르겠다. 튜토리얼이 아닌 진짜 실무에서 쓸수 있는 Airflow 시스템을 구축 하려고 한다. 반면 읽어도 큰 도움이 안 될 분들은 이런 분들 입니다. 대규모 작업이 필요하며, 빠..

airflow에서 dag을 만들고 실행을 시켰는데 task는 분명히 뜨고 해당 task도 실행중으로 나오는데, 상태값만 변하고 실행되지 않는 경우가 있습니다. 다음과 같은 경우가 그 현상에 해당 합니다. 아래와 같이 Task의 상태값이 분명히 실행으로 변경되었는데, 아래의 네모칸에 실행중(초록색)으로 변경되지 않습니다. 아무리 기다려도 역시 그대로 입니다. 그래서 어떻게 할지 고민했는데, airflow에서 강제로 "Run"을 시킬 수 있습니다. 강제로 queue에 넣어주는 방법입니다. 가끔씩 should가 느리게 동작해서 반응이 없을 경우가 있는데 이렇게 강제로 Run을 시켜서 동작 시키는 경우가 많기 때문에 같은 증상으로 보고 똑같이 실행해 봤습니다. 다음과 같이 강제로 "Run"으로 변경 해봤습니다...

docker 컨테이너를 삭제하려고 하는데, 계속해서 Removal In Progress으로 상태가 멈춰 있고 지워지지 않는 현상을 발견하였습니다. 그래서 찾아봤는데 다음과 같은 방법으로 해결 하면 된다고 하여 확인하였는데... 무엇을 지워야 할지 몰라서 당황하였습니다. 참고 URL github.com/moby/moby/issues/22312 Containers stuck in "Removal In Progress" · Issue #22312 · moby/moby Output of docker version: Client: Version: 1.11.0 API version: 1.23 Go version: go1.5.4 Git commit: 4dc5990 Built: Wed Apr 13 18:34:23 20..

Airflow를 세팅하고 CPU를 보면 아무것도 하지 않고 있는데, CPU를 40~50% 정도 점유하는 경우가 있습니다. 하지만 이는 몇 가지 옵션만 변경한다면 쉽게 CPU 점유율을 낮출 수 있습니다. 또한 운영상에 큰 지장이 없습니다. 저의 경우 GCP 위에서 docker를 통해서 airflow를 운영하고 있으며, 옵션은 worker,scheduler,webserver 이렇게 3개의 컨테이너에 적용하였습니다. 모두 동일하게 적용하였습니다. 1 2 3 4 5 6 7 8 9 10 11 # The scheduler constantly tries to trigger new tasks (look at the # scheduler section in the docs for more information). Thi..

Airflow는 ETL스케줄링 오픈소스로 많은 분들이 사용하고 있습니다. 하지만 오픈소스라서 여러가지 불편점이 있는데요. 그 중 알람의 경우도 그렇습니다. 물론 Dag에 slack 알람을 받을 수 있도록 설정이 가능 한데요. 이 경우 매우 불편하게 하나씩 설정해야 하는 경우가 있습니다. 그래서 Airflow의 postgreSQL에 있는 데이터를 기반으로 slack 알람을 받을 수 있는 python 스크립트를 만들어봤습니다. 우선 세팅이 필요한데요. slack으로 알람을 받기 위해서는 2가지가 필요합니다. slack channel ID slack webhook url 우선 channel ID의 경우 다음과 같이 받을 수 있습니다. slack을 웹으로 접속 할 경우 위에 url을 확인 할 수 있는데요. 다음..

embulk plugin에서 embulk-output-bigquery부분에서 아래와 같은 오류가 날 때 해결 방법에 대해서 적어 봤습니다. Incompatible table partitioning specification when copying to the column partitioned table 위의 오류는 mysql -> big query로 갈 때 big query 쪽에 이미 partition이 되어 있는 경우 발생하였습니다. 상황에 따라서 다양하게 발생하는 것으로 보이는데 저의 경우 위의 상황으로 발생하였습니다. 관련해서 embulk plugin 쪽에 내용을 보면 다음과 같이 표기되어 있습니다. 아래의 내용은 mode에서 replace를 사용할 때 관련된 내용인 거 같은데, 사실 해결 방법과 별..

환경 - OS : Mac OS - Input : mysql (AWS) - Output : big query (GCP) 오류 내용 org.embulk.exec.PartialExecutionException: java.lang.RuntimeException: java.sql.SQLException: The server time zone value 'KST' is unrecognized or represents more than one time zone. You must configure either the server or JDBC driver (via the serverTimezone configuration property) to use a more specifc time zone value if you..

이 테스트 시나리오는 sql-server to sql-server입니다. 다른 시나리오와 맞지 않을 수 있으니 꼭 참고 부탁 드립니다. Ⅰ. Embulk 프로세스 1. Embulk를 실행 시 적용된 쿼리를 읽어서 하나의 Transaction으로 만듭니다.2. 생성 된 Transaction을 각각의 task로 쪼개는 작업을 합니다.이때 테이블이 Thread 개수에 맞게 생성되는 것을 볼 수 있습니다.3. 해당 Thread 개수 만큼의 테이블을 생성하여, 각각 테이블에 parallel 처리로 데이터를 Insert 시킵니다.4. Temp Table에 데이터를 모두 넣으면 이것을 모두 Union 하여, 최종 목적지 테이블에 insert into 시킵니다. 위에 프로세스를 보면 CPU 성능이 매우 중요합니다. 그..

Embulk를 요즘 ETL 툴로 많이 사용하고 있는데, 계속 설치하다가 오류가 나서, 여기저기 찾아보았으나 안되서 삽질 하루종일 하다가 참 쉽게(?) 방법을 알아내서 글을 올립니다. 환경은 다음과 같이 구축하려고 하였습니다. - Embulk를 Windows Server에 설치하여, 사용하고자 함 보통 Embulk는 Linux에서 사용하는 것으로 알고 있습니다. ( Plugin 이 호환이 안되는 경우가 많다고...)하지만 이미 Windows Server에 구축이 되어있는 상황이므로 어쩔수 없이 써야 할 경우였기 때문에 Windows Server에 설치를 시도하였습니다. 그런데, 다음과 같은 오류가 발생 합니다. 인터넷에 찾아보았으나, 역시 해결되지 않았습니다. 그래서 다음과 같이 설치하였습니다. Step ..