강사님 *** !!!! Please make sure that all your Airflow components (e.g. schedulers, webservers, workers and triggerer) have the same 'secret_key' configured in 'webserver' section and time is synchronized on all your machines (for example with ntpd) See more at https://airflow.apache.org/docs/apache-airflow/stable/configurations-ref.html#secret-key *** Could not read served logs: 403 Client Error: FORBIDDEN for url: http://e1efbc97ae25:8793/log/dag_id=dags_email_operator/run_id=manual__2024-08-24T06:22:20.118495+00:00/task_id=send_email_task/attempt=1.log [2024-08-24, 06:22:25 UTC] {local_task_job_ runner.py:123 } ▶ Pre task execution logs [2024-08-24, 06:22:25 UTC] {warnings.py:112} WARNING - /home/***/.local/lib/python3.12/site-packages/***/utils/email.py:155: RemovedInAirflow3Warning: Fetching SMTP credentials from configuration variables will be deprecated in a future release. Please set credentials using a connection instead. send_mime_email(e_from=mail_from, e_to=recipients, mime_msg=msg, conn_id=conn_id, dryrun=dryrun) [2024-08-24, 06:27:49 UTC] {taskinstance.py:3301} ERROR - Task failed with exception 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/operators/email.py", line 79, in execute send_email( File "/home/airflow/.local/lib/python3.12/site-packages/airflow/utils/email.py", line 80, in send_email return backend( ^^^^^^^^ File "/home/airflow/.local/lib/python3.12/site-packages/airflow/utils/email.py", line 155, in send_email_smtp send_mime_email(e_from=mail_from, e_to=recipients, mime_msg=msg, conn_id=conn_id, dryrun=dryrun) File "/home/airflow/.local/lib/python3.12/site-packages/airflow/utils/email.py", line 280, in send_mime_email smtp_conn.starttls() File "/usr/local/lib/python3.12/smtplib.py", line 779, in starttls self.sock = context.wrap_socket(self.sock, ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/ssl.py", line 455, in wrap_socket return self.sslsocket_class._create( ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/ssl.py", line 1042, in create self.do handshake() File "/usr/local/lib/python3.12/ssl.py", line 1320, in do_handshake self._sslobj.do_handshake() TimeoutError: _ssl.c:983: The handshake operation timed out 위 문구와 함께 현재 smtp.gmail.com 에 접속이 안되는데.. (worker에서도 접속이 안됨) 혹시 다른 설정이 필요할게 있을까요?
안녕하세요, 인강 수강후 현업에서 Airflow 를 통해서 Dag 을 수행하고 있는데요, 서비스 PM 등의 이유로 수시간(3~5시간) 정도 pause 를 진행하고 다시 unpause 를 하게 되는 경우가 있습니다. 이럴때 unpasue 를 클릭하게 된다면 비록 Dag 에 catchup = false 로 지정해 놓더라도, 가장 최근에 수행 되었어야 할 Dag Run 은 수행되게 되는데요 Ex ) schedule_interval = 10,20,30 1,2,3 * * * * Dag.catchup = False pause = 01:15 unpause = 03:25 수행되는 Dag run 의 data_interval_end = 03:20 그런데 여기서 3시 20 분 dag_run 을 수행시키지 않기 위해서 생각을 하다가, Dag 의 default_args 의 start_date 값을 datetime.now 로 주는것에 대해서 생각을 해봤는데요, 당장 간단하게 테스트를 했을때는 큰 문제가 없었는데, 혹시 해당 케이스가 문제가 되는 케이스가 있을까요? 그리고 start_date 를 now 로 하는것 말고도 다른 방안이 있다면 좋은 방안 부탁 드립니다 감사합니다!
안녕하세요. EmailOperator 실습과정에서 오류가 발생해서 질문을 남깁니다! 설정을 한 후 실행을 하니. OSError: [Errno 99] Cannot assign requested address 가 발생해서 도움이 필요합니다..! docker-compose.yaml 파일에 이렇게 설정을 하였고, 에러는 다음과 같습니다!
안녕하세요. 8장에서 postgres 에 insert하는 dag를 실행하면 아래와 같은 오류로 대기하다가 실패로 끝납니다. 구글에 찾아보면 'secret_key'를 모두 동일하게 적용해주라고 하는데, 어떻게 적용해야 되는지 찾지는 못해서 이렇게 문의하게 되었습니다. 아래 소스도 git에 올라와 있는 소스 그대로 올려서 수행했기 때문에 따로 올리지는 않았습니다. docker-compose.yaml dags_python_with_postgres 위에 상태에서 최종 오류로 떨어지면 오류 메세지를 확인할 수 있었습니다 .ㅜ 사실 원인은 위에 메세지가 아닌것 같아요. ERROR - Failed to execute job 1119 for task insrt_postgres (connection to server at "172" (0.0.0.172), port 5432 failed: Connection timed out Is the server running on that host and accepting TCP/IP connections? connection to server at "28.0.3" (28.0.0.3), port 5432 failed: Connection timed out Is the server running on that host and accepting TCP/IP connections? 그리고, 저의 docker-compose.yaml 에서 postgres_custom, postgres의 설정부분입니다. 그리고, airflow의 connection 연결할 때도 포트는 5432로 했습니다. 사실 질문답변을 검색해보면 저와 비슷한 오류가 1개 있어서 확인을 해봤는데요 그 오류는 5429에서 5432로 변경해서 해결됐다고 나온 것 같고 저는 5432인데 해결이 안돼서 제가 실수한게 있는지 확인부탁드립니다. ㅜ
아래와 같이 dag 작성하고 smtp 설정을 해주었는데 오류가 발생했습니다. RemovedInAirflow3Warning: Fetching SMTP credentials from configuration variables will be deprecated in a future release. Please set credentials using a connection instead. send_mime_email(e_from=mail_from, e_to=recipients, mime_msg=msg, conn_id=conn_id, dryrun=dryrun) [2024-07-22, 06:53:07 UTC] {configuration.py:1053} WARNING - section/key [smtp/smtp_user] not found in config [2024-07-22, 06:53:07 UTC] {email.py:271} INFO - Email alerting: attempt 1 오류내용은 위와같습니다!
안녕하세요! 강사님 서울시 공공데이터 API를 이용해 데이터를 추출하는 것을 보고 저도 넥슨 오픈 API를 가지고 데이터를 출력해보려고 DAG과 PythonOperator를 만들어봤는데요, 보통 민감정보는 .env파일에다 적어놓고 가져와서 사용하기에 이렇게 짰는데, .env파일은 gitignore에 들어가니 dag실행시 읽어올 수가 없는데 이런 경우는 어떻게 할까요...? 강의에서 보여주신 것처럼 서울시 공공 api에서 simplehttpoperator를 사용하고, airflow웹에서 variable 변수를 사용했듯이 이것도 이렇게 해야할까요..?!
안녕하세요! 강사님 질문이 하나 있는데요, 제가 중간에 docker를 삭제했다가 다시 설치해서 하고 있는데 다른 것들은 잘 되는데, 서울시 공공 데이터 API를 가지고 Custome Operator를 하는 과정에서 dags_seoul_api_ corona.py 이 실행이 안되어서 문의드려요! 처음에 directory가 없다는 오류가 떠서 생성하는 코드를 추가해봤는데, 추가한 이후에 dag 실행해도 실패해서 로그를 확인해보니 permission 거부 문제로 오류가 발생한 것을 확인하였습니다. docker-compose.yaml에서 volumes는 잘 설정되었는데, 혹시 제가 AIRFLOW_CORE_EXECUTOR를 LocalExecutor로 바꾼 게 문제가 되었을까요? databricks와 airflow를 연동하면서 dns를 설정하고, command를 좀 수정하였는데 이 부분이 문제였을까요...? ChatGPT에 물어보니 Webserver로 들어가서 Docker 컨테이너 내에서 이 디렉토리의 소유자를 airflow 사용자로 변경하면 된다고 하는데, [sudo] password for default: 부분에서 비밀번호 에러가 나네요...! 저는 따로 default에 대한 비밀번호를 설정하지 않아서 해당 비밀번호를 잘 모르는 상태입니다..! docker를 재설치해야할 것 같은데... 하기 전에 강사님께 문의 드립니다! 편하게 답변 주시면 감사하겠습니다!!
안녕하세요, 수업을 대부분 수강하고 실제 현업에서 사용중에 있는데 문의사항이 있어서 질문 드립니다. 현재 상황은 이렇습니다. DAG 구성 - 5분단위 스케줄링 -4개의 task - task1 >> task2 >> task3 >> task4 - 각 task 별 timeout =5분 문제 상황은 task 2번이 한달에 한번씩 data 가 많아지면 5분까지 타임아웃이 걸릴때가 있는 것인데요, 이때 그다음 Dag run 이 수행되면서 task 2 번이 동시에 수행 되는 시간이 조금 있는데 그때 데이터 처리가 중복으로 처리되는 현상이 발생하게 됩니다. 그래서 가능하면 task2 을 동시에 돌리는걸 막고 싶었는데요, 처음 생각해낸 방법은 task_concurrency 옵션을 task 에 주어서 1개만 돌수 있게 바꾸고 timeout 을 조금더 넉넉하게 주려고 했으나, 만에하나 해당 task 가 10분이상 걸린다면 dag run 이 수행되고있는것 제외 2개가 더 웨이팅을 하는것이 되고, 이게 누적이 될수도 있을것으로 보여서 문제로 인지 했습니다. 서비스 적으로 5분내에 돌수 있게 하거나, 아니면 5분 스케줄링을 변경하는 방법을 고려해야 하지만 해당 고려 없이 혹시 airflow 단에서 할수 있는 작업이 있을까요? ex . runninng 중인 task 와 대기중인 task 가 하나정도 있다면 해당 task 는 스킵하는 옵션 등입니다..
airflow와 postgres 간의 connection 오류 문제입니다. airflow UI -> admin-> connections에서 postgres 연결설정 docker-compose.yaml 설정 dag 코드 입력 airflow tasks test postgres_loader execute_sql_query 2023-01-01 시에 오류가 뜹니다ㅠ [2024-06-21T15:40:45.514+0900] { dagbag.py:545 } INFO - Filling up the DagBag from /home/kim/airflow/dags [2024-06-21T15:40:45.805+0900] { taskinstance.py:2076 } INFO - Dependencies all met for dep_context=non-requeueable deps ti=<TaskInstance: postgres_loader.execute_sql_query __airflow_temporary_run_2024-06-21T06:40:45.755970+00:00__ [None]> [2024-06-21T15:40:45.811+0900] { taskinstance.py:2076 } INFO - Dependencies all met for dep_context=requeueable deps ti=<TaskInstance: postgres_loader.execute_sql_query __airflow_temporary_run_2024-06-21T06:40:45.755970+00:00__ [None]> [2024-06-21T15:40:45.812+0900] { taskinstance.py:2306 } INFO - Starting attempt 1 of 1 [2024-06-21T15:40:45.812+0900] { taskinstance.py:2388 } WARNING - cannot record queued_duration for task execute_sql_query because previous state change time has not been saved [2024-06-21T15:40:45.813+0900] { taskinstance.py:2330 } INFO - Executing <Task(PostgresOperator): execute_sql_query> on 2023-01-01 00:00:00+00:00 [2024-06-21T15:40:45.855+0900] { taskinstance.py:2648 } INFO - Exporting env vars: AIRFLOW_CTX_DAG_OWNER='airflow' AIRFLOW_CTX_DAG_ID='postgres_loader' AIRFLOW_CTX_TASK_ID='execute_sql_query' AIRFLOW_CTX_EXECUTION_DATE='2023-01-01T00:00:00+00:00' AIRFLOW_CTX_TRY_NUMBER='1' AIRFLOW_CTX_DAG_RUN_ID='__airflow_temporary_run_2024-06-21T06:40:45.755970+00:00__' [2024-06-21T15:40:45.858+0900] { taskinstance.py:430 } INFO - ::endgroup:: [2024-06-21T15:40:45.870+0900] { sql.py:276 } INFO - Executing: INSERT INTO sample_table (key, value) VALUES ('hello', 'world') [2024-06-21T15:40:45.875+0900] { taskinstance.py:441 } INFO - ::group::Post task execution logs [2024-06-21T15:40:45.875+0900] { taskinstance.py:2905 } ERROR - Task failed with exception Traceback (most recent call last): File "/home/kim/.local/lib/python3.10/site-packages/airflow/models/ taskinstance.py ", line 465, in _execute_task result = _execute_callable(context=context, **execute_callable_kwargs) File "/home/kim/.local/lib/python3.10/site-packages/airflow/models/ taskinstance.py ", line 432, in _execute_callable return execute_callable(context=context, **execute_callable_kwargs) File "/home/kim/.local/lib/python3.10/site-packages/airflow/models/ baseoperator.py ", line 401, in wrapper return func(self, *args, **kwargs) File "/home/kim/.local/lib/python3.10/site-packages/airflow/providers/common/sql/operators/ sql.py ", line 277, in execute hook = self.get_db_hook() File "/home/kim/.local/lib/python3.10/site-packages/airflow/providers/common/sql/operators/ sql.py ", line 188, in get_db_hook return self._hook File "/usr/lib/python3.10/ functools.py ", line 981, in __get__ val = self.func(instance) File "/home/kim/.local/lib/python3.10/site-packages/airflow/providers/common/sql/operators/ sql.py ", line 150, in _hook conn = BaseHook.get_connection(conn_id) File "/home/kim/.local/lib/python3.10/site-packages/airflow/hooks/ base.py ", line 83, in get_connection conn = Connection.get_connection_from_secrets(conn_id) File "/home/kim/.local/lib/python3.10/site-packages/airflow/models/ connection.py ", line 519, in get_connection_from_secrets raise AirflowNotFoundException(f"The conn_id `{conn_id}` isn't defined") airflow.exceptions.AirflowNotFoundException: The conn_id `my_postgres_connection` isn't defined
선생님이 4:21초에 from airflow.sensors.sql import SqlSensor 는 provider에 있는게 아니라 core에 있는 sensor라고 알려주셨는데 airflow 버전 2.9.1에서는 SqlSensor 가 apache-airflow-providers-common-sql 패키지에 포함되어 있다고 하는데 그러면 버전 2.9.1에서는 airflow core에 있는 sensor를 사용하지 못하는 건가요??
안녕하세요 실습중에 오류 발생해서 질문드립니다. dag파일 잘못 작성하여 dag import error가 발생한 경우, dag파일을 수정하고 적용시키기 위해서는 매번 docker compose down 후, compose up으로 재기동 해줘야하나요? 간단한 오탈자 수정하는데도 재기동이 필요한지 궁금합니다.
안녕하세요, wsl을 이용한 linux설치 강의를 따라하고 있었는데 시작버튼을 눌러 powershell 을 검색하면 windows powershell ISE 만 나오고 windows powershell 은 나오지 않습니다. 별도의 windows powershell 설치가 필요할까요? 구글링을 해봐도 아직 방법을 못찾았기에 질문드립니다. 어떻게 진행하면 될까요 ?
선생님 안녕하세요! 덕분에 Airflow에 대해 깊이 있게 공부하고 있습니다! 감사합니다. 다름이 아니라 Airflow를 잘 쓰고자 하는 마음에 질문드립니다! 제가 지금 구축해야하는 환경이 Google Cloud 기반에서 DataLake와 Warehouse를 구축 해야 하는 상황에서 Airflow 강의를 참고해 도입 예정에 있습니다. 사 내 인프라 팀은 잘 갖춰져 있으나 데이터팀은 아직 미약한 상태에요.. 구글링 해서 살펴보았을 때 Airflow의 전처리의 대부분이 BigQuery의 SQL을 통해 원하는 데이터를 가져와 전처리하는 로직으로 구성되어 있는거 같더라구요. 저는 Pandas라는 라이브러리가 익숙한 것도 있고 SQL 쿼리로 관리하기보다 Pandas 코드로 관리하고자하는 마음에 Airflow와 Pandas의 조합은 어떻게 쓰면 좋다라거나 참고 블로그에 대해 알고싶고 또 선생님 조언을 들어보고 싶습니다 ㅠㅠ 아직 Airflow를 완전히 이해하지 못했지만 걱정되는 점은 Pandas 사용 시 데이터를 읽었을 때 인메모리에 많은 양의 데이터가 올라가 주의하지 않으면 구축하려는 Cloud Composer의 스펙이 오버될거 같은 느낌이 들어서요.. 또 다른 궁금한 점은 전처리 구간이 많을 수록 BigQuery에 저장하면서 불러들이는 식으로 작업하시는 지도 궁금합니다!! 장애 발생 시 어떤 구간에서 발생했으며 Retry 시 저장하면서 가야 정확한 에러 구간에 대해 모니터링이 가능해보여서요.. 마지막으로.. dags를 관리하는 아키텍쳐? 방안에 대해서 유행하거나 픽스된 방법론이 있는 지도 궁금해요 백엔드의 디자인패턴과 유사한.. 질문이 많죠.. 백엔드하다 데이터 엔지니어 업무가 처음이다 보니 궁금한게 많네요.. 다시 정리를 하면 질문은 아래와 같습니다. 긴 글 읽어주셔서 감사합니다 ( _ _ ) Airflow와 Pandas 조합을 사용하고자 할 때 선생님의 조언이 궁금합니다. 전처리 구간 마다 생기는 View Table이 데이터 양이 많을 때 저장하는 지 궁금합니다. git에서 dags를 관리하는 방법론이 궁금합니다.
안녕하세요! 에어플로우를 사용하고 싶어서 강의를 수강하게 되었는데 수업을 너무 잘 해주셔서 많은 도움을 받고 있습니다. 다름이 아니라 제가 개인적인 공부를 목적으로 개인 블로그를 하고 있는데, 에어플로우에 관해 강의 자료를 토대로 정리해서 게시물을 작성해도 될지.. 여쭤보고 싶습니다..! 강의 자료를 그대로 사용하지는 않고 내용을 바탕으로 정리해서 작성할 계획입니다..!
