[feat] 일간 공업 용수 수질 DAG (ETL) 작성

[secureWKkim]

• 한 번 받아둔 정수장 코드 데이터 어떻게 쓸지 고민하기. 이걸 dag로 다시 돌릴 필요는 없으니, 서진님이 법정동 코드 load해서 쓴 것처럼 쓰던지..
• 정수장 코드를 변수로 활용해 파일명 및 csv 내 value로 넣기. 모두 받아 오도록.
• json to csv <- 안됐던 이유는 응답 데이터 크기가 너무 컸기 때문. numOfRows=100으로 하면 된다. + json key 문제.
• pageNo를 변수로 만들거나, 에러 반환 전까지 순회 => 후자는 문제가 있어 전자로 성공!
• dataframe 스키마 변경
• 정수장 이름을 키로 정수장 코드 & 주소 간 join.
• dag full 작동 local test
• 로컬 변경 사항 원상복귀 혹은 폐기 처분 후 origin/main pull merge 후 충돌 날 것들 미리 처분
• gcs to bigquery operator 도입
• 현재 시준 기점 6개월 치 데이터 살펴보기.
• csv 일별 생성 말고 하나로 모아버리기(실시간인데도 양이 너무 적어서)
• string을 datetime으로 인식하게끔 코드 변경 후, 그래도 안되면 빅쿼리 안에서 sql 쿼리 날려서 변경
• historical data backfill DAG 하나 더 만들기??

[secureWKkim]

index must be int, not str 에러

아마 타입 변환이 제대로 안된것 아닌가 싶다. params 등 뭔가가 dict가 아닌 list로 인식되고 있다던지.. 그래서 디버거 돌려보면서 생각해보기로 했다.
=>해결. csv 첫 줄인 스키마를 갖고 오면서 생기는 문제였다.

[secureWKkim]

현재 문제

num_of_pages=1인 경우, 가장 큰 initial df에다가 응답df+주소df(series) 결합한 새 df를 append 하는 가공 작업을 해줘야 하는데, 그걸 안해줘서 문제.

그래서 for i in range(2, num_of_pages + 1):로 돼있던 곳에서 숫자 2를 1로 바꾸고, if i>1: 이란 조건문 추가해 줌.

반복되는 로직을 함수로 빼야 하는데, 그게 무엇이고 필요한 인자는 뭔가.