파이프라인 구축 시 고려할 점
환경 설정 config
✔ github에 리포지토리 만들고 내 작업환경인 VSC에서 로컬 폴더 열고 연동시켜줌.
✔ .gitignore 파일 생성해서 불필요한 파일은 git에서 제외해줌.
✔ API 데이터를 가져오고, dataframe을 활용하기 위해 requests와 pandas 각각 설치 완료.
✔ Docker 환경 설정 docker-compose.yml 작성 및 컨테이너 띄우기까지 완료
지난 경험에서 로컬에서 작업할 때는 독립적인 환경을 확보하기 위해서 python 가상환경을 활용하기로 했기 때문에 가상환경 활성화해줌. bash에서는 그 커맨드가 아래와 같다. activate 하면 (venv)가 뜨고, 그럼 잘 실행된 것.
source venv/Scripts/activate
deactivate
requests, pandas 등의 라이브러리도 다운로드 받음.
docker의 경우에는 계속 로컬환경에서 써왔으므로, docker desktop 실행하고 잘 실행되는 것 확인함.
docker에서 사용할 Airflow, Spark, BigQuery 등을 연동시키는 docker-compose.yml을 생성함.
그리고 이후에, BigQuery 만으로는 로컬 테스트가 좀 불편하고, 또 BigQuery로 최종 적재하기 전 데이터 검증을 하기 위해서 로컬용 PostgreSQL도 추가해줌.
docker-compose up -d 엄청 오래 걸리네. 이미지들 다 다운받느라 엄청 걸림. 각각의 초기 설정이 이미 내부적으로 진행되어서 더욱 느린 것도 있다고 함. 이 이후에는 로컬에 도커 이미지가 캐싱되어 있으므로, 바로 실행됨.
spark UI가 안 뜨는 문제가 있어서 해결해줌.
docker-compose.yml에 spark-master > environment에 확실히 WEBUI PORT에 대해 명시해주는 게 중요했음.
그리고 docker에 데이터가 마운트될 때, 자꾸만 알 수 없는 에러가 떴는데, 바로 마운트될 때
C:\Program Files\Git이 prefix가 자꾸만 느닷없이 붙는다는 것. 찾아보니 이게 window 기반에서 docker 돌릴 때 생기는 버그라고 함. 이를 해결하기 위해, 도커 위에서 export MSYS_NO_PATHCONV=1 시전해줌.
기본적으로 commit convention은 아래와 같이 따랐다.
뿐만 아니라, 이 과정 전체에서 gitflow의 흐름을 따랐다. 그 과정에서 자주 쓰는 코드들은 다음과 같음.
또한 데이터 파이프라인을 구축하면서, gitflow 흐름을 따라 가면서 테스트를 꾸준히 진행해줌.
이후에 CI/CD 구축 단계에서
feature 단위의 테스트, 각 컴포넌트들이 잘 되는지 즉 integration test, 그리고 수집된 데이터가 유효한 값인지 확인하는 Data Quality Test, End-to-end test 등을 구축해줌.
데이터 수집 Extract
✔ 사용할 주식 데이터 원천 선택 완료 - Yahoo
✔ Alpha Vantage 가입하고 OAuth 인증을 위해 무료 API 키 발급받아서, .env 파일에 저장.
✔ 결과적으로 과금 이슈 때문에 Yahoo Finance로 바꾸게 됨.
✔ 개발 단계에서는 로컬 파일 저장소에 저장하는 방식으로 구현함.
Yahoo와 Alpha Vantage API 중 OAuth 경험을 위해 우선은 Alpha Vantage로 정함. 공식 API기 때문에 안정적이라는 이유도 한몫함.
일단 데이터 가져올 때까지는 Docker와 관계없기 때문에 가상환경 켜고 코드 짬.
OAuth 처리: 내 개인 API key 받아왔고, 따옴표 없이 .env에 저장해줌. python은 자동으로 string으로 처리하기 때문.
그리고 env 파일을 활용하기 위해서 python-dotenv 다운받아줌.
그러나 결과적으로 하루 호출 제한 때문에 Alpha가 아닌 Yahoo를 쓰기로 함.
Yahoo로 코드 바꿔서 주식 정보값들 가져오는 데에 어렵지 않게 성공.
1차적으로 가져오는 것은 원본 데이터를 그대로 유지하면서 데이터 손실을 방지해야 하므로 저 상태로 저장되는 것을 확인. 나중에 완성도를 위해 Storage를 활용하는 식으로 하겠지만, 개발 단계에서는 로컬 파일 시스템을 활용.
결과적으로는 Yahoo에서 가져오기로 함...
그런데 branch에서 기껏 다 만들고 나서 branch를 병합 안 하고 제거해버렸다... 그걸 되돌리느라 git을 다시 다뤄보고 좋았다 ㅋㅋ 식겁했는데 그냥 단순했다, 휴.
git reflog
일단은 지난 1일치가 아닌 5일치 과거 데이터들을 스크래핑으로 가져옴.
데이터 저장 GCS로 - I, Extract 직후
내 버킷:
우선 유저 본인이 아니라 앱, VM, 또는 기타 서비스가 GCP 리소스에 액세스할 수 있도록 권한을 부여해줘야 함.
그게 서비스 계정. 사람이 아닌 프로그램이 사용하도록 설계된 계정. 그걸 연동할 수 있게 함.
JSON 형식의 KEY file을 통해 권한 인증 후 리소스에 대한 접근이 가능함.
컨테이너 위에 올려주고, env에도 설정해줌. docker-compose.yml도 설정해줌.
그리고 도커에도 gcs 관련 라이브러리 설치해주고.
.env 파일 적용 및 GOOGLE_APPLICATION_CREDENTIALS=$ 이거 써서 내부에서 KEY 쓸 수 있게하고.
./keys: /opt/keys 마운트해서 모든 컨테이너에서 GCS key 파일 접근 가능.
아래 코드로 테스트 해줌.
이때 /bin/bash 가 없어서 걍 sh로 대체해줌.
인고의 과정 끝에, 다시 short-term 이라는 버킷을 만들고, 내 스파크를 저장소 관리자로 권한을 줌.
권한 문제 때문에 두세시간 씨름하고 결국 해냄.
결과적으로 첫번째 short-term 버킷의 short_term/collected 내에 저장됨! 껄껄.
이 과정에서 Dockerfile과 docker-compose.yml의 차이가 궁금해짐...
데이터 처리 processing
GCS에서 값을 가져오게 하려면, Spark가 GCS에서 데이터를 읽어야 하고 그러려면 추가 설정이 필요함.
이 부분이 또 매우 귀찮음. hadoop 환경설정 파일부터 시작해서 별걸 다 가져와서 Dockerfile 수정하고 등등.
결과적으로 GCS에서 값 불러와서 GCS로 값 저장하는 과정까지 성공.
이때 인풋은 csv, 아웃풋은 parquet로 지정했다.
* 그런데 본질적으로, Spark 에서의 데이터 처리와 (가공) BigQuery에서의 데이터 처리의 역할 분담을 어떻게 해야할지에 대해 고민해봐야 할 듯.
GCS 에서 파일 가져오고, 파일 쓰고 하는 그 과정에서 매우 귀찮은 일들이 있었는데 그것들 모두 Dockerfile과 docker-compose에 시전해줌.
데이터 적재 Load
BigQuery로 적재하게끔 바꾸는 중....
GCS temp storage가 있어야 된다고 해서 그거 path 관련 아주 쌩난리를 치고 어찌저찌 BigQuery까지 접근을 했는데,
BigQuery에서도 허용이 필요해서 허용해줌. 그리고 새로운 data set 만들어줬고.
빅쿼리에 적재됐는지 확인하기
bq --project_id=fluid-mix-452511-j3 query --use_legacy_sql=false 'SELECT * FROM `fluid-mix-452511-j3.short_term.sp500_top50` LIMIT 10;'
BigQuery 에서의 최적화
-> python으로 스크립트를 짜줌.
Looker
Ver 2
최적화
데이터 fetching 단에서의 최적화 전:
멀티스레딩 적용을 통해 54.7584s => 14.4796s로 73.5% 단축.
데이터 변환 최적화
'데이터 엔지니어링' 카테고리의 다른 글
| [SQLD] SQL 스킬을 업그레이드하자! - 자격증은 덤이지 (0) | 2025.03.09 |
|---|---|
| [데이터 엔지니어링] 데이터 웨어하우스, 레이크, 마트와 데이터 모델링 💫 (0) | 2025.02.24 |
| [Spark] 캡스톤 프로젝트🚧: from HIVE to Kafka (0) | 2025.02.23 |
| [Spark] A급 💯 데이터 엔지니어를 위한 Aggregations & Joins (0) | 2025.02.22 |
| [Spark] Spark의 꽃🌺, Transformations (0) | 2025.02.21 |
