[2025-04-07T14:51:04.313+0000] {taskinstance.py:3310} ERROR - Task failed with exception Traceback (most recent call last): File "/opt/airflow/dags/yfinance_to_snowflake_ fullrefresh.py ", line 56, in load raise e File "/opt/airflow/dags/yfinance_to_snowflake_ fullrefresh.py ", line 41, in load cur.execute(f"USE SCHEMA {schema};") File "/home/airflow/.local/lib/python3.12/site-packages/snowflake/connector/ cursor.py ", line 1087, in execute Error.errorhandler_wrapper(self.connection, self, error_class, errvalue) File "/home/airflow/.local/lib/python3.12/site-packages/snowflake/connector/ errors.py ", line 284, in errorhandler_wrapper handed_over = Error.hand_to_other_handler( ^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/home/airflow/.local/lib/python3.12/site-packages/snowflake/connector/ errors.py ", line 339, in hand_to_other_handler cursor.errorhandler(connection, cursor, error_class, error_value) File "/home/airflow/.local/lib/python3.12/site-packages/snowflake/connector/ errors.py ", line 215, in default_errorhandler raise error_class( snowflake.connector.errors.ProgrammingError: 002043 (02000): 01bb881b-0000-c602-0000-cde100062082: SQL compilation error: Object does not exist, or operation cannot be performed. During handling of the above exception, another exception occurred: Traceback (most recent call last): File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/ taskinstance.py ", line 767, in execute task result = execute callable(context=context, **execute_callable_kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/ taskinstance.py ", line 733, in execute callable return ExecutionCallableRunner( ^^^^^^^^^^^^^^^^^^^^^^^^ File "/home/airflow/.local/lib/python3.12/site-packages/airflow/utils/operator_ helpers.py ", line 252, in run return self.func(*args, **kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/ baseoperator.py ", line 406, in wrapper return func(self, args, *kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/home/airflow/.local/lib/python3.12/site-packages/airflow/decorators/ base.py ", line 266, in execute return_value = super().execute(context) ^^^^^^^^^^^^^^^^^^^^^^^^ File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/ baseoperator.py ", line 406, in wrapper return func(self, args, *kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/home/airflow/.local/lib/python3.12/site-packages/airflow/operators/ python.py ", line 238, in execute return_value = self.execute_callable() ^^^^^^^^^^^^^^^^^^^^^^^ File "/home/airflow/.local/lib/python3.12/site-packages/airflow/operators/ python.py ", line 256, in execute_callable return runner.run (*self.op_args, **self.op_kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/home/airflow/.local/lib/python3.12/site-packages/airflow/utils/operator_ helpers.py ", line 252, in run return self.func(*args, **kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/airflow/dags/yfinance_to_snowflake_ fullrefresh.py ", line 62, in load cur.execute(f"REMOVE {table_stage}/{file_name}") File "/home/airflow/.local/lib/python3.12/site-packages/snowflake/connector/ cursor.py ", line 1087, in execute Error.errorhandler_wrapper(self.connection, self, error_class, errvalue) File "/home/airflow/.local/lib/python3.12/site-packages/snowflake/connector/ errors.py ", line 284, in errorhandler_wrapper handed_over = Error.hand_to_other_handler( ^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/home/airflow/.local/lib/python3.12/site-packages/snowflake/connector/ errors.py ", line 339, in hand_to_other_handler cursor.errorhandler(connection, cursor, error_class, error_value) File "/home/airflow/.local/lib/python3.12/site-packages/snowflake/connector/ errors.py ", line 215, in default_errorhandler raise error_class( snowflake.connector.errors.ProgrammingError: 090105 (22000): 01bb881b-0000-c619-0000-cde10006306e: Cannot perform STAGE RM. This session does not have a current database. Call 'USE DATABASE', or use a qualified name. Snowflake 세션에서 데이터베이스 또는 스키마가 제대로 설정되지 않아 발생한 문제로 보인다는데;;; 뭔가 실행에 이상이 있는건지;;;;
강사님 안녕하세요. airflow의 rest api를 이용하여 task를 재실행하는 코드를 개발하는 중 궁금한 점이 생겨서 문의드립니다. (참고로 실행환경은 회사pc입니다) ~/v1/dags/{dag_id}/dagRuns/{execution_date}/taskInstances/{task_id}/clear 위와같이 특정 task의 실행 시간으로 실행하면 404에러가 뜨고 "detail": "The requested URL was not found on the server. If you entered the URL manually please check your spelling and try again.", "status": 404, "title": "Not Found", "type": "about:blank" REST-API문서를 살펴보니 clearTaskInstances가 있어서 ~/api/v1/dags/{dag_id}/clearTaskInstances 이렇게 조회하면 405 에러가 뜹니다. "detail": "The method is not allowed for the requested URL.", "status": 405, "title": "Method Not Allowed", "type": "about:blank" /clear이든 /clearTaskInstances 둘다 작동하지 않은데, 방법이 있을까요? 두번째의 경우 서버에서 작동을 막는것일까요?
airflow dags list를 하면 yfinance_to_snowflake_fullrefresh.py 에서 no module named helpers 에러가 떴습니다. pip3 install helpers 를 통해 설치를 했는데도, 동일한 에러가 발생하고 있습니다. (python은 homebrew 통해 설치했고, homebrew 설치 시에 패키지 전역 설치가 안된다는 문제가 있었는데, 이는 해결해서 helpers Install은 완료 했습니다.)
1. Data Drift 발생 시에 머신러닝 모델이 동작하지 않게 될 것이라는 것은 서비스는 돌아가지만, 머신러닝의 모델이 원래 기대했던 성능을 내지 못할 것을 의미하는 것일까요? 예를 들어 추천 모델을 돌리고 있다면, 추천이 제대로 되지 않고 엉뚱한 것을 추천한다는 것을 의미하는 것인가요? 2. 주기적으로 데이터의 분포를 점검하는 필요가 있다면 어느 정도 주기여야 할까요? 3. 데이터의 분포가 어느 정도로 변하게 되면 이상 신호로 받아 들이게 되는 것인가요? 평균의 변화가 아닌 분산의 변화만으로도 이상 신호로 보아야 하는 것인지요? (뭔가 더 복잡하면 데이터의 성격에 따라 이러한 분포의 변화에 대해 반응해야 하는 수준이 다 다를 것으로도 느껴지기도 합니다.)
안녕하세요 강사님 airflow config를 변경하기 위해 시도해보았는데 이렇게 expose를 false->true로 변경하여도 airflow web에서는 변경되지 않아서 혹시 어디가 잘못 되었는지 문의드립니다 그리고 dag인식이 느린 것 같아 dag_dir_list 값을 100으로 변경하였는데 cpu만 괜찮다면 다른 변수들은 수정할 필요가 없을까요?
CountryCapital_v3.py 실습과정을 진행중에 갑자기 airflow UI 에서 DAGs 리스트들이 나오지 않는 현상을 겪고 있습니다. airflow dags list 명령어에 대한 결과이고, 도커에서 airflow 도 잘 띄우고 있습니다. 이전까지는 잘 되었는데 다음 사진과 같이 dag들이 갑자기 나오지 않고 있습니다.
안녕하세요 강사님 혹시 vsc에서 airflow관련 함수들의 자동완성을 제공하는 익스텐션이 따로 있나요? 아니면 코드를 원래 전체 작성해야 하나요? 위 그림에 ti.xcom_pull의 경우 ti. 하고 자동완성을 하려고 봤더니 제공하는 함수가 없어서 질문드립니다 근데 뒤에 task_ids는 자동완성이 되네요
섹션 3에 있는 Snowflake 환경 익히기 실습에서 nps.csv 파일은 어디에 있나요? 환경 익히기 실습에서 진행하는 google colab 파일과 깃헙에 있는 country_capital_to snowflake.py 파일의 차이를 알고 싶습니다. 최종적으로 우리가 airflow를 사용하기 위해 스노우플레이크 내에 만들어져 있어야하는 스키마, 테이블 및 적재되어야하는 csv파일 정리해서 알려주시면 감사드리겠습니다.
Airflow Variables 설정시 스노우플레이크와 연결을 위한 snowflake_userid, snowflake_password, snowflake_account 설정은 이해를 합니다. 그런데 첫번째인 Country_capital_url 왜 설정하는지 이해가 안갑니다.
안녕하세요 선생님, 강의를 정말 재밌게 수강하고 있습니다. 섹션 8 PostgresToSnowflake DAG를 실행해서 Snowflake에 user_session_channel 테이블과 session_timestamp 테이블에 CSV 파일 로드할 때 오류가 있는것 같아요. 간단하게 해결할 수 있는 문제인데 DAG 파이썬 파일에서 os import 구문이 빠져있는거 같아요 import os 이거 넣어서 간단하게 해결했습니다! 좋은 강의 만들어주셔서 감사하고 곧 추가될 강의도 기다리고 있겠습니다!
강의를 들으면서 데이터 엔지니어를 준비하고 있는 학부생입니다. 취업사이트에 데이터 엔지니어를 찾아보면 대게 경력 위주로 채용을 진행하고 있는데 데이터 엔지니어가 신입으로 일을 할 수 있는 방법이 있는 지 ? 아니면 신입으로서 바로 데이터 엔지니어로 일하기는 조금 어려운게 현실인지 궁금하여 글 작성합니다. 괜찮으시다면 데이터엔지니어 직무 입사자가 이런 경험이 있으면 채용하겠다는 부분이 있는 지 알려주시면 감사하겠습니다.
안녕하세요, 데이터 엔지니어에 관심을 가지고 이제 막 공부를 시작한 학생입니다. 섹션 8 yfinance 첫 번째 강의를 수강하던 도중에 에러가 생겨서 문의 남깁니다. 작성되어 있는 코드 중에 extract(symbol) >> load(file_path, symbol, schema, table) 이 부분에서 문제가 생겼었습니다. 그래서 file_path = extract(symbol) extract(symbol) >> load(file_path, schema, table) 이렇게 수정해서 오류를 해결했는데, airflow web ui에서 다른 오류가 발생했길래 또 찾아보니 load함수에서 문제가 생긴 것 같습니다. file_path를 load함수에서도 다시 만들어줘서 snowflake에 데이터를 적재할 때 올바른 파일 경로를 찾지 못해서 생기는 오류가 발생하는거 같아요. def load(file_path, schema, table): cur = util.return_snowflake_conn("snowflake_conn") try: cur.execute(f"USE SCHEMA {schema};") cur.execute( f"""CREATE TABLE IF NOT EXISTS {table} ( date date, open float, close float, high float, low float, volume int, symbol varchar )""" ) cur.execute("BEGIN;") delete_sql = f"DELETE FROM {table}" logging.info(delete_sql) cur.execute(delete_sql) # 바로 받은 file_path 사용 (더 이상 경로 재생성 X) util.populate_table_via_stage(cur, table, file_path) cur.execute("COMMIT;") except Exception as e: cur.execute("ROLLBACK;") raise e finally: file_name = os.path.basename(file_path) table_stage = f"@%{table}" cur.execute(f"REMOVE {table_stage}/{file_name}") cur.close() 요렇게 변경해주니까 데모 영상에 나온 것과 같이 똑같이 스노우플레이크에 잘 적재되었습니다.! 제가 제대로 해결한게 맞을까요??
이 과정은 google colab을 사용하여 ETL 프로세스를 만들어 스노우플레이크에 적재하는 방식을 보여줍니다. 두 개의 컬럼을 갖고 있는 country_capital.csv 파일을 텍스트로 풀어 쓰고 콤마로 나눠 country와 capital 로 나누는 과정을 설명해주는데, 이 과정을 진행하는 이유가 단순 궁금합니다. 스노우플레이크 GUI 환경에서 add data 하여 스테이지-테이블 순으로 적재를 하면 되는데 코드화 하여 적재하는 구분하여 적재하는 이유가 단순히 궁금합니다. 혹시, 나중에 처리하여 올리기 힘든 데이터의 경우 이렇게 전처리 과정을 미리 거쳐 올리는 방법을 알려주시는건가요?
토근 발급하는 과정에서 해당 오류로 다음 진도를 못나가고 있습니다..ㅜㅜ 이전 서울시 공공데이터 실습하는 과정에서 requests 정상 import 및 문제없이 작동하였는데, 이번 실습에선 module 'requests' has no attribute 'post' 오류가 발생하네요. - 오류를 해결하기 위해 requests.py 파일이 있는지 확인 - requests 정상 import 확인 - requests reinstall 진행하였으나 문제 해결이 되지 않았습니다..혹시 놓친 부분이 있는지 도와주세요. 감사합니다! import requests client_id = '' redirect_url = 'https://example.com/oauth' authorize_code = '' token_url = 'https://kauth.kakao.com/oauth/token' data = { 'grant_type': 'authorization_code', 'client_id': client_id, 'redirect_url': redirect_url, 'code': authorize_code } response = requests.post(token_url, data=data) tokens = response.json() print(tokens) 추가로 아래 AI 답변 관련하여 확인 해보았습니다. 그럼에도 놓친 사항이 있을 수 있으니 함께 확인하여 답변해주시면 감사하겠습니다!
