ããã«ã¡ã¯ã ä»å¹Ž4æã« ãšãã°ã¢ ã«å
¥ç€ŸããããŒã¿ãšã³ãžãã¢ã®è°·å
ã§ãã ãã®èšäºã¯ Enigmo Advent Calendar 2020 ã® 20æ¥ ç®ã®èšäºã§ãã ç®æ¬¡ ã¯ããã« ããããåæåŠçãšã¯ïŒ Airflowã«ããåæåŠç æ€èšŒæã®ã³ãŒã ãµã³ãã«ãAirflowç»é¢ã§èŠããšïŒ åæé
å»¶ãªãæã®Airflowãã° åæé
å»¶ããæã®Airflowãã° åæã¿ã€ã ã¢ãŠãæã®Airflowãã° ææ æåŸã« ã¯ããã« ã³ããçŠã®äžãçããã©ã®ããã«ãéããã§ããããã ç§ã¯ãªã¢ãŒãã¯ãŒã¯ãç¶ããŠãŸãããèªå®
ã®ãªã¢ãŒããã¹ã¯ã¯ãŒã¯ç°å¢ãããã«æŽããªãã£ããããèãã¯ãã·ã§ã³ã§åº§ããªããããŒããŒãã«ã§3ã¶æçµéããé ã«èº«äœã®ç¯ã
ã§æ²é³ŽããããŸãããç«èãå é... ããŠã ãšãã°ã¢ ã§ã®ä»äºãå幎以äžçµéããããŒã¿åæåºç€ã®éçºéçšä¿å®ãBIäžã§ã®ããŒã¿æŽåãªã©ã察å¿ããããŠããã ããŠãŸãããä»åã¯ç€Ÿå
ã§å©çšãããŠããAirflowã® åæåŠç ã®è©±ãããããšæããŸãã å°ããã¡ãã®èšäºãæžãã«ããããåç°å¢ã®éå»èšäº Enigmo Advent Calendar 2018 ããããŸãã®ã§ãè¯ããã°ãã¡ããã芧ãã ããã ããããåæåŠçãšã¯ïŒ äŸãã°ã以äžã®ãããªããŒã¿åæåºç€äžã§ã®ETLåŠçããã£ããšããŸãã åŠç1. 飿ºå
DBã®ããããŒãã«ãDataLake(GCS)ãžãã¡ã€ã«ãšããŠæ ŒçŽåŸãDWH(BigQuery)ãžããŒããã åŠç2. DWHäžã®ããŒã¿ãå å·¥åŸãå€éšãžCSV飿ºãã äžèšã®åŠçã¯é çªã«å®è¡ãããªããšå¿
èŠãªããŒã¿ãå«ããæœåºãã§ããªããªã£ãŠããŸããå€éšãžããŒã¿é£æºã§ããªããªããŸãããã®ããã«ã¯åŠç1ã®æ£åžžçµäºç¢ºèªåŸã«åŠç2ã宿œãããšãããåæåŠçãæèããå®è£
ãããå¿
èŠããããŸãã äŸã ãšã·ã³ãã«ã§ãããæ¥åäžã§ã¯åŸã
ã«ETLåŠçãå¢ããŠãããè€éåããŠãããŸãã ããããåæåŠçã¯ãåãã¯ãªããŠãåé¡ã«ãªããªãããšãå€ãã®ã§ãããåŸã
ã«åŠçé
å»¶ãçºçããŠã¿ã€ãã³ã°ãåããªããªã£ãããäºæãã¬äžéšã®åŠçãšã©ãŒãåå ã§é¢é£ããåŸç¶åŠçãå
šãŠæå³ãã¬ç¶æ
ã§åãåºããŠããŸã£ããããŸãããããªããªãããã«ããèšèšæç¹ã§åæã«ã€ããŠæèããããšã倧äºã ãšæããŸãã Airflowã«ããåæåŠç Airflowã§ã¯ ExternalTaskSensor ã䜿çšããããšã§å®è£
å¯èœãšãªããŸãã Airflowã®ãœãŒã¹äžã§ã¯dependãšããçšèªã䜿ãããŠããã®ã§ãAirflowã®äžçã§ã¯ãåæãã§ã¯ãªãããäŸåããšåŒãã æ¹ãè¯ãã®ãããããŸãããé©åŒèªã¿æ¿ããŠããã ããã°... ã§ã¯ãAirflowã§ã®åææ€èšŒçšãµã³ãã«ã³ãŒããäœæããŠã¿ãŸããã®ã§ãå®éã«åãããªããæ€èšŒããããšæããŸããå°ãAirflowããŒãžã§ã³ã¯1.10.10ãšãªããŸãã å
ã»ã©ã®ETLåŠçãäŸã«ããŠãäžèšã®å
å®¹ã§æ€èšŒããŠã¿ãŸãã å®éã®ãã¡ã€ã«åºåãDBãžã®ããŒãåŠçãªã©ã¯ããã§ã¯å²æããŠã DummyOperator ã«çœ®ãæããŠãŸãã åŠç1. 飿ºå
DBã®ããããŒãã«ãDataLake(GCS)ãžãã¡ã€ã«ãšããŠæ ŒçŽåŸãDWH(BigQuery)ãžããŒããã dag_id: sample_db_to_dwh_daily schedule: æ¥æ¬¡16:00 tables: TABLE_DAILY_1, TABLE_DAILY_2, TABLE_DAILY_3 åŠç2. DWHäžã®ããŒãã«ãçšããŠããŒã¿å å·¥åŸãå€éšããŒã«ãžCSV飿ºãã dag_id: sample_dwh_to_file_daily_for_sync_daily_16 schedule: æ¥æ¬¡17:00 tables: TABLE_DAILY_1, TABLE_DAILY_2, TABLE_DAILY_3 æ€èšŒæã®ã³ãŒã ãŸãã¯åŠç1ã®ãµã³ãã«ã³ãŒãã«ãªããŸãã from airflow.models import DAG from datetime import datetime from airflow.operators.dummy_operator import DummyOperator dag = DAG( dag_id='sample_db_to_dwh_daily', start_date=datetime(2020, 12, 1), schedule_interval='0 16 * * *' ) tables = ['TABLE_DAILY_1', 'TABLE_DAILY_2', 'TABLE_DAILY_3' ] for table_name in tables: db_to_dwh_operator = DummyOperator( task_id='db_to_dwh_operator_%s' % table_name, dag=dag ) terminal_operator = DummyOperator( task_id='terminal_operator_%s' % table_name, dag=dag, trigger_rule='none_failed' ) db_to_dwh_operator >> terminal_operator 次ã«åŠç2ã®ãµã³ãã«ã³ãŒãã§ãã from airflow.models import DAG from airflow.operators.dummy_operator import DummyOperator from airflow.sensors.external_task_sensor import ExternalTaskSensor from airflow.utils.state import State from datetime import datetime, timedelta tables = ['TABLE_DAILY_1', 'TABLE_DAILY_2', 'TABLE_DAILY_3' ] def _dwh_to_file_operator(dag, table_name, depend_external_dag_id, depend_external_dag_execution_delta): dwh_to_file_operator = DummyOperator( task_id='dwh_to_file_operator_%s' % table_name, dag=dag ) sensors = [] sensors.append(ExternalTaskSensor( task_id='wait_for_sync_db_to_dwh_%s' % table_name, external_dag_id=depend_external_dag_id, external_task_id='terminal_operator_%s' % table_name, execution_delta=depend_external_dag_execution_delta, allowed_states=[State.SUCCESS, State.SKIPPED], dag=dag)) for sensor in sensors: sensor >> dwh_to_file_operator terminal_operator = DummyOperator( task_id='terminal_operator_%s' % table_name, dag=dag, trigger_rule='none_failed') dwh_to_file_operator >> terminal_operator args = { 'start_date': datetime(2020, 12, 3) } dag = DAG( dag_id='sample_dwh_to_file_daily_for_sync_daily', default_args=args, schedule_interval='0 17 * * *' ) for table_name in tables: _dwh_to_file_operator( dag=dag, table_name=table_name, depend_external_dag_id='sample_db_to_dwh_daily', depend_external_dag_execution_delta=timedelta(hours=1) ) ãµã³ãã«ãAirflowç»é¢ã§èŠããšïŒ äžèš2ã€ã®DAGãAirflowã®WebUIã§èŠããšä»¥äžã®ããã«ãªããŸãã åæåŠçãããoperatorã¯ãwait_for_sync_db_to_dwh ãã§ããå³ã§ã ExternalTaskSensor ã®ã¢ã€ã³ã³ã«ãªã£ãŠãŸããã ä»åã¯3ããŒãã«ããããããã䞊ååŠçãæ³å®ããŠããããã3ã€ãããŸãã ãããŠãããããã該åœããããŒãã«ã®terminal_operatoråŠçå®äºã確èªåŸãåŸç¶åŠçãå®è¡ãããŸãã ã§ã¯äžèšã®åæåŠçãAirflowã®ãã°ã§ã©ã®ããã«è¡šç€ºããããèŠãŠãããããšæããŸãã åæé
å»¶ãªãæã®Airflowãã° ãŸãã¯ãæ¢ã« sample_db_to_dwh_daily ãå®äºããŠããå Žåã§ãã 3ããŒãã«æ¯ã®ãã°ãèŒããŠãããŸãã TABLE_DAILY_1 ~~ DAG: sample_dwh_to_file_daily_for_sync_daily Task Instance: wait_for_sync_db_to_dwh_TABLE_DAILY_1 2020-12-07 17:00:00 ~~ [2020-12-09 02:00:09,833] {taskinstance.py:669} INFO - Dependencies all met for <TaskInstance: sample_dwh_to_file_daily_for_sync_daily.wait_for_sync_db_to_dwh_TABLE_DAILY_1 2020-12-07T17:00:00+00:00 [queued]> [2020-12-09 02:00:09,840] {taskinstance.py:669} INFO - Dependencies all met for <TaskInstance: sample_dwh_to_file_daily_for_sync_daily.wait_for_sync_db_to_dwh_TABLE_DAILY_1 2020-12-07T17:00:00+00:00 [queued]> [2020-12-09 02:00:09,841] {taskinstance.py:879} INFO - -------------------------------------------------------------------------------- [2020-12-09 02:00:09,841] {taskinstance.py:880} INFO - Starting attempt 1 of 1 [2020-12-09 02:00:09,841] {taskinstance.py:881} INFO - -------------------------------------------------------------------------------- [2020-12-09 02:00:09,848] {taskinstance.py:900} INFO - Executing <Task(ExternalTaskSensor): wait_for_sync_db_to_dwh_TABLE_DAILY_1> on 2020-12-07T17:00:00+00:00 [2020-12-09 02:00:09,851] {standard_task_runner.py:53} INFO - Started process 48063 to run task [2020-12-09 02:00:09,905] {logging_mixin.py:112} INFO - Running %s on host %s <TaskInstance: sample_dwh_to_file_daily_for_sync_daily.wait_for_sync_db_to_dwh_TABLE_DAILY_1 2020-12-07T17:00:00+00:00 [running]> ip-123-456-78-910.dokoka_toku.compute.internal [2020-12-09 02:00:09,919] {external_task_sensor.py:117} INFO - Poking for sample_db_to_dwh_daily.terminal_operator_TABLE_DAILY_1 on 2020-12-07T16:00:00+00:00 ... [2020-12-09 02:00:09,921] {base_sensor_operator.py:123} INFO - Success criteria met. Exiting. [2020-12-09 02:00:09,924] {taskinstance.py:1065} INFO - Marking task as SUCCESS.dag_id=sample_dwh_to_file_daily_for_sync_daily, task_id=wait_for_sync_db_to_dwh_TABLE_DAILY_1, execution_date=20201207T170000, start_date=20201208T170009, end_date=20201208T170009 [2020-12-09 02:00:19,834] {logging_mixin.py:112} INFO - [2020-12-09 02:00:19,834] {local_task_job.py:103} INFO - Task exited with return code 0 TABLE_DAILY_2 ~~ DAG: sample_dwh_to_file_daily_for_sync_daily Task Instance: wait_for_sync_db_to_dwh_TABLE_DAILY_2 2020-12-07 17:00:00 ~~ [2020-12-09 02:00:09,831] {taskinstance.py:669} INFO - Dependencies all met for <TaskInstance: sample_dwh_to_file_daily_for_sync_daily.wait_for_sync_db_to_dwh_TABLE_DAILY_2 2020-12-07T17:00:00+00:00 [queued]> [2020-12-09 02:00:09,839] {taskinstance.py:669} INFO - Dependencies all met for <TaskInstance: sample_dwh_to_file_daily_for_sync_daily.wait_for_sync_db_to_dwh_TABLE_DAILY_2 2020-12-07T17:00:00+00:00 [queued]> [2020-12-09 02:00:09,839] {taskinstance.py:879} INFO - -------------------------------------------------------------------------------- [2020-12-09 02:00:09,840] {taskinstance.py:880} INFO - Starting attempt 1 of 1 [2020-12-09 02:00:09,840] {taskinstance.py:881} INFO - -------------------------------------------------------------------------------- [2020-12-09 02:00:09,846] {taskinstance.py:900} INFO - Executing <Task(ExternalTaskSensor): wait_for_sync_db_to_dwh_TABLE_DAILY_2> on 2020-12-07T17:00:00+00:00 [2020-12-09 02:00:09,848] {standard_task_runner.py:53} INFO - Started process 48062 to run task [2020-12-09 02:00:09,902] {logging_mixin.py:112} INFO - Running %s on host %s <TaskInstance: sample_dwh_to_file_daily_for_sync_daily.wait_for_sync_db_to_dwh_TABLE_DAILY_2 2020-12-07T17:00:00+00:00 [running]> ip-123-456-78-910.dokoka_toku.compute.internal [2020-12-09 02:00:09,916] {external_task_sensor.py:117} INFO - Poking for sample_db_to_dwh_daily.terminal_operator_TABLE_DAILY_2 on 2020-12-07T16:00:00+00:00 ... [2020-12-09 02:00:09,918] {base_sensor_operator.py:123} INFO - Success criteria met. Exiting. [2020-12-09 02:00:09,921] {taskinstance.py:1065} INFO - Marking task as SUCCESS.dag_id=sample_dwh_to_file_daily_for_sync_daily, task_id=wait_for_sync_db_to_dwh_TABLE_DAILY_2, execution_date=20201207T170000, start_date=20201208T170009, end_date=20201208T170009 [2020-12-09 02:00:19,832] {logging_mixin.py:112} INFO - [2020-12-09 02:00:19,832] {local_task_job.py:103} INFO - Task exited with return code 0 TABLE_DAILY_3 ~~ DAG: sample_dwh_to_file_daily_for_sync_daily Task Instance: wait_for_sync_db_to_dwh_TABLE_DAILY_3 2020-12-07 17:00:00 ~~ [2020-12-09 02:00:09,829] {taskinstance.py:669} INFO - Dependencies all met for <TaskInstance: sample_dwh_to_file_daily_for_sync_daily.wait_for_sync_db_to_dwh_TABLE_DAILY_3 2020-12-07T17:00:00+00:00 [queued]> [2020-12-09 02:00:09,837] {taskinstance.py:669} INFO - Dependencies all met for <TaskInstance: sample_dwh_to_file_daily_for_sync_daily.wait_for_sync_db_to_dwh_TABLE_DAILY_3 2020-12-07T17:00:00+00:00 [queued]> [2020-12-09 02:00:09,837] {taskinstance.py:879} INFO - -------------------------------------------------------------------------------- [2020-12-09 02:00:09,837] {taskinstance.py:880} INFO - Starting attempt 1 of 1 [2020-12-09 02:00:09,837] {taskinstance.py:881} INFO - -------------------------------------------------------------------------------- [2020-12-09 02:00:09,844] {taskinstance.py:900} INFO - Executing <Task(ExternalTaskSensor): wait_for_sync_db_to_dwh_TABLE_DAILY_3> on 2020-12-07T17:00:00+00:00 [2020-12-09 02:00:09,847] {standard_task_runner.py:53} INFO - Started process 48061 to run task [2020-12-09 02:00:09,901] {logging_mixin.py:112} INFO - Running %s on host %s <TaskInstance: sample_dwh_to_file_daily_for_sync_daily.wait_for_sync_db_to_dwh_TABLE_DAILY_3 2020-12-07T17:00:00+00:00 [running]> ip-123-456-78-910.dokoka_toku.compute.internal [2020-12-09 02:00:09,915] {external_task_sensor.py:117} INFO - Poking for sample_db_to_dwh_daily.terminal_operator_TABLE_DAILY_3 on 2020-12-07T16:00:00+00:00 ... [2020-12-09 02:00:09,917] {base_sensor_operator.py:123} INFO - Success criteria met. Exiting. [2020-12-09 02:00:09,920] {taskinstance.py:1065} INFO - Marking task as SUCCESS.dag_id=sample_dwh_to_file_daily_for_sync_daily, task_id=wait_for_sync_db_to_dwh_TABLE_DAILY_3, execution_date=20201207T170000, start_date=20201208T170009, end_date=20201208T170009 [2020-12-09 02:00:19,831] {logging_mixin.py:112} INFO - [2020-12-09 02:00:19,831] {local_task_job.py:103} INFO - Task exited with return code 0 泚ç®ããŠæ¬²ããè¡ã¯äžèšãšãªããŸãã â» ä»ããŒãã«ã®ãã°ãåãã§ãã®ã§ãããããå
ã¯èšèŒã岿ããŸã [2020-12-09 02:00:09,919] {external_task_sensor.py:117} INFO - Poking for sample_db_to_dwh_daily.terminal_operator_TABLE_DAILY_1 on 2020-12-07T16:00:00+00:00 ... ãããåæåŠçã®ãã°ã«ãªããŸãã ä»åã¯æ¢ã«æ£åžžçµäºããŠããããã1åã®ç¢ºèªããè¡ãããŠãããããã®åŸãåŸç¶åŠçãå«ããŠæ£åžžçµäºããŠãããšãåãããŸãã åæé
å»¶ããæã®Airflowã㰠次ã«åæã®ã¿ã¹ã¯ãé
å»¶ããŠæ£åžžçµäºããå Žåã®Airflowãã°ãã¿ãŠãããããšæããŸãã ~~~ DAG: sample_dwh_to_file_daily_for_sync_daily Task Instance: wait_for_sync_db_to_dwh_TABLE_DAILY_1 2020-12-08 17:00:00 ~~~ [2020-12-10 02:00:04,174] {taskinstance.py:669} INFO - Dependencies all met for <TaskInstance: sample_dwh_to_file_daily_for_sync_daily.wait_for_sync_db_to_dwh_TABLE_DAILY_1 2020-12-08T17:00:00+00:00 [queued]> [2020-12-10 02:00:04,182] {taskinstance.py:669} INFO - Dependencies all met for <TaskInstance: sample_dwh_to_file_daily_for_sync_daily.wait_for_sync_db_to_dwh_TABLE_DAILY_1 2020-12-08T17:00:00+00:00 [queued]> [2020-12-10 02:00:04,182] {taskinstance.py:879} INFO - -------------------------------------------------------------------------------- [2020-12-10 02:00:04,182] {taskinstance.py:880} INFO - Starting attempt 1 of 1 [2020-12-10 02:00:04,182] {taskinstance.py:881} INFO - -------------------------------------------------------------------------------- [2020-12-10 02:00:04,189] {taskinstance.py:900} INFO - Executing <Task(ExternalTaskSensor): wait_for_sync_db_to_dwh_TABLE_DAILY_1> on 2020-12-08T17:00:00+00:00 [2020-12-10 02:00:04,192] {standard_task_runner.py:53} INFO - Started process 21683 to run task [2020-12-10 02:00:04,245] {logging_mixin.py:112} INFO - Running %s on host %s <TaskInstance: sample_dwh_to_file_daily_for_sync_daily.wait_for_sync_db_to_dwh_TABLE_DAILY_1 2020-12-08T17:00:00+00:00 [running]> ip-123-456-78-910.dokoka_toku.compute.internal [2020-12-10 02:00:04,259] {external_task_sensor.py:117} INFO - Poking for sample_db_to_dwh_daily.terminal_operator_TABLE_DAILY_1 on 2020-12-08T16:00:00+00:00 ... [2020-12-10 02:01:04,322] {external_task_sensor.py:117} INFO - Poking for sample_db_to_dwh_daily.terminal_operator_TABLE_DAILY_1 on 2020-12-08T16:00:00+00:00 ... [2020-12-10 02:02:04,385] {external_task_sensor.py:117} INFO - Poking for sample_db_to_dwh_daily.terminal_operator_TABLE_DAILY_1 on 2020-12-08T16:00:00+00:00 ... [2020-12-10 02:03:04,448] {external_task_sensor.py:117} INFO - Poking for sample_db_to_dwh_daily.terminal_operator_TABLE_DAILY_1 on 2020-12-08T16:00:00+00:00 ... [2020-12-10 02:04:04,511] {external_task_sensor.py:117} INFO - Poking for sample_db_to_dwh_daily.terminal_operator_TABLE_DAILY_1 on 2020-12-08T16:00:00+00:00 ... (çç¥) [2020-12-10 10:27:29,163] {external_task_sensor.py:117} INFO - Poking for sample_db_to_dwh_daily.terminal_operator_TABLE_DAILY_1 on 2020-12-08T16:00:00+00:00 ... [2020-12-10 10:28:29,226] {external_task_sensor.py:117} INFO - Poking for sample_db_to_dwh_daily.terminal_operator_TABLE_DAILY_1 on 2020-12-08T16:00:00+00:00 ... [2020-12-10 10:29:29,273] {external_task_sensor.py:117} INFO - Poking for sample_db_to_dwh_daily.terminal_operator_TABLE_DAILY_1 on 2020-12-08T16:00:00+00:00 ... [2020-12-10 10:30:29,298] {external_task_sensor.py:117} INFO - Poking for sample_db_to_dwh_daily.terminal_operator_TABLE_DAILY_1 on 2020-12-08T16:00:00+00:00 ... [2020-12-10 10:31:29,360] {external_task_sensor.py:117} INFO - Poking for sample_db_to_dwh_daily.terminal_operator_TABLE_DAILY_1 on 2020-12-08T16:00:00+00:00 ... [2020-12-10 10:31:29,363] {base_sensor_operator.py:123} INFO - Success criteria met. Exiting. [2020-12-10 10:31:29,366] {taskinstance.py:1065} INFO - Marking task as SUCCESS.dag_id=sample_dwh_to_file_daily_for_sync_daily, task_id=wait_for_sync_db_to_dwh_TABLE_DAILY_1, execution_date=20201208T170000, start_date=20201209T170004, end_date=20201210T013129 [2020-12-10 10:31:33,698] {logging_mixin.py:112} INFO - [2020-12-10 10:31:33,698] {local_task_job.py:103} INFO - Task exited with return code 0 å
ã»ã©ã®Pokingè¡ããã°ã«è¡šç€ºããç¶ããŠããããšãåãããŸããä»åã¯éäžã§ã¿ã¹ã¯ãæåã§åãããŠæ£åžžçµäºãããŸããã å°ãPokingã®ééã¯ExternalTaskSensorã®åŒæ°ã§ poke_interval ãèšå®ãããšå€æŽãã§ããŸããããœãŒã¹ãèŠããšããã©ã«ãã¯60ç§ã®ããã§ãããã°ãšãäžèŽããŠãŸããã åæ ã¿ã€ã ã¢ãŠã æã®Airflowãã° æ¥åã§ã¯éæ©èœèŠä»¶ã倧äºã ãšæããŸãã Airlfowã§ã¯ãExternalTaskSensorã®åŒæ°ã§ timeout ãããããã§ãã ãã¡ãã§ ã¿ã€ã ã¢ãŠã (default: 1é±éã§ããããé·ã£)ãé©åã«èšå®ããŠããšã©ãŒæ€ç¥ãããŠã¿ãããšæããŸãã sample_dwh_to_file_daily_for_sync_hourly/wait_for_sync_db_to_dwh_TABLE_HOURLY_1/2020-12-09T17:00:00+00:00/4.log. [2020-12-11 18:21:14,539] {taskinstance.py:669} INFO - Dependencies all met for <TaskInstance: sample_dwh_to_file_daily_for_sync_hourly.wait_for_sync_db_to_dwh_TABLE_HOURLY_1 2020-12-09T17:00:00+00:00 [queued]> [2020-12-11 18:21:14,551] {taskinstance.py:669} INFO - Dependencies all met for <TaskInstance: sample_dwh_to_file_daily_for_sync_hourly.wait_for_sync_db_to_dwh_TABLE_HOURLY_1 2020-12-09T17:00:00+00:00 [queued]> [2020-12-11 18:21:14,551] {taskinstance.py:879} INFO - -------------------------------------------------------------------------------- [2020-12-11 18:21:14,551] {taskinstance.py:880} INFO - Starting attempt 4 of 9 [2020-12-11 18:21:14,552] {taskinstance.py:881} INFO - -------------------------------------------------------------------------------- [2020-12-11 18:21:14,563] {taskinstance.py:900} INFO - Executing <Task(ExternalTaskSensor): wait_for_sync_db_to_dwh_TABLE_HOURLY_1> on 2020-12-09T17:00:00+00:00 [2020-12-11 18:21:14,568] {standard_task_runner.py:53} INFO - Started process 84130 to run task [2020-12-11 18:21:14,625] {logging_mixin.py:112} INFO - Running %s on host %s <TaskInstance: sample_dwh_to_file_daily_for_sync_hourly.wait_for_sync_db_to_dwh_TABLE_HOURLY_1 2020-12-09T17:00:00+00:00 [running]> ip-123-456-78-910.dokoka_toku.compute.internal [2020-12-11 18:21:14,639] {external_task_sensor.py:117} INFO - Poking for sample_db_to_dwh_hourly.terminal_operator_TABLE_HOURLY_1 on 2020-12-10T16:00:00+00:00 ... [2020-12-11 18:22:14,701] {external_task_sensor.py:117} INFO - Poking for sample_db_to_dwh_hourly.terminal_operator_TABLE_HOURLY_1 on 2020-12-10T16:00:00+00:00 ... [2020-12-11 18:23:14,764] {external_task_sensor.py:117} INFO - Poking for sample_db_to_dwh_hourly.terminal_operator_TABLE_HOURLY_1 on 2020-12-10T16:00:00+00:00 ... [2020-12-11 18:24:14,828] {external_task_sensor.py:117} INFO - Poking for sample_db_to_dwh_hourly.terminal_operator_TABLE_HOURLY_1 on 2020-12-10T16:00:00+00:00 ... [2020-12-11 18:25:14,892] {external_task_sensor.py:117} INFO - Poking for sample_db_to_dwh_hourly.terminal_operator_TABLE_HOURLY_1 on 2020-12-10T16:00:00+00:00 ... [2020-12-11 18:25:14,898] {taskinstance.py:1145} ERROR - Snap. Time is OUT. Traceback (most recent call last): File "/opt/airflow/venv/lib/python2.7/site-packages/airflow/models/taskinstance.py", line 983, in _run_raw_task result = task_copy.execute(context=context) File "/opt/airflow/venv/lib/python2.7/site-packages/airflow/sensors/base_sensor_operator.py", line 116, in execute raise AirflowSensorTimeout('Snap. Time is OUT.') AirflowSensorTimeout: Snap. Time is OUT. [2020-12-11 18:25:14,899] {taskinstance.py:1202} INFO - Marking task as FAILED.dag_id=sample_dwh_to_file_daily_for_sync_hourly, task_id=wait_for_sync_db_to_dwh_TABLE_HOURLY_1, execution_date=20201209T170000, start_date=20201211T092114, end_date=20201211T092514 [2020-12-11 18:25:15,070] {logging_mixin.py:112} INFO - [2020-12-11 18:25:15,070] {local_task_job.py:103} INFO - Task exited with return code 1 hourlyã«å€æŽããŠå®æœãããã°ã§ããããSnap. Time is OUT.ããšè¯ãæã(?)ã«ãšã©ãŒãšãªã£ãŠãããŸããã æ¥åäžã§ã¯ããç®ã«ããããããªããã°å
容ã§ãã... ä»ã«ãè²ã
ããããã§ããããã®èŸºããæããŠããã°åºæ¬çãªäœ¿ãæ¹ã¯æããããšã«ãªãããšæããŸãã ææ å®éã«AirflowåæåŠçããã£ãŠã¿ãŠæã£ãã®ã§ããã åæãåã察象ã®operatorãã©ãã«ããã åæãåãoperatorã®æå»å·®ã¯ã©ãã ãããã éæ©èœèŠä»¶ã«ãã£ã ã¿ã€ã ã¢ãŠã èšå®ã§é©åã«ãšã©ãŒã§èœãšãã è€éã«ãªããšDAGéã®é¢ä¿æ§ããããã°ã©ããWebUIã§ãµã¯ããšã¿ãã ããæ°ã«ãªããŸããã ãåæãåã察象ã®operatorãã¯ãterminal_operator_<ããŒãã«å>ããšããŸããã ãããŒåŠçãªã®ã§åé·ã«ã¯ãªã£ãŠããŸããŸãããå䞊ååŠçæ¯ã«å ããŠãããã»ããDAGä¿®æ£æã®åæåœ±é¿ãæèããªããŠè¯ããªãã®ããªãšæã£ãããã§ãã ãŸããDummyOperatorå©çšæã« ãtrigger_rule='none_failed'ãã®åŒæ°ãä»ãå ããªããšãå
ã«å®è¡ãããã±ãŒã¹ããããŸããã®ã§æ³šæãå¿
èŠããã§ãã ãåæãåãoperatorã®æå»å·®ãã§ãããäœåºŠãç¹°ãè¿ããŠããšæ··ä¹±ããŠããŸãå¯èœæ§ãããã®ã§ããã¹ãã§ãååã«æ°ãã€ããŠå¯Ÿå¿ããªããšãããªãã§ãã(toèªå)ãå®éã®æ¥åã§ãtimedeltaã§æå®ããæå»ã«èª€ãããã¹ãã§èŠèœãšããŠããŸãããã®çµæããªãªãŒã¹åŸã«åæåŠçãæ³å®æééãã«çµãããé
å»¶ããŠããŸããå€éšé£æºã®ã¿ã€ãã³ã°ã«éã«åããªããªãåé¡ã«ãªã£ãŠããŸããŸãã...ã ä»åã®æ€èšŒäŸã¯ããäºãdailyå®è¡ã ã£ãã®ã§ãããé »åºŠãç°ãªããšæ
éãªå¯Ÿå¿ãæ±ãããããã§ãã ãã ããã®èŸºã®å¶åŸ¡ã¯Airflowã®ã³ã¢ãªå¶åŸ¡ãªã®ã§ãä»åŸãå©çšè
ãæèããªãã§æžãããã«dag_idãšãªãã·ã§ã³åŒæ°ã§æž¡ããããåæãåã£ãŠããããããªæ©èœããããšè¯ããªãšãæããŸãããå©çšé »åºŠãå€ããªããšããããã£ãå®è£
ãæ€èšããªããšãããªãã®ãããããŸããã ããšã¯ãWebUIäžã«ãŠDAGã®é¢ä¿æ§ãã°ã©ãã§èŠãããšãæå³ããoperatorã§åæãåããŠãããã®ç¢ºèªãããããããšæããŸããã èŠæ¹ãåãããªãã ããªã®ããæªå®è£
ãªã®ãææ¡ã§ããŠãªãã®ã§ãããWebUIãã¿ãéãä»åã®Airflowã®ããŒãžã§ã³1.10.10ã§ã¯ãªãããã§ãã æåŸã« Airflowã§ã®åæåŠçã«ã€ããŠå°ãã§ãäŒããã°ãšæãããã®ããŒãã§èšäºãæžããŠã¿ãŸããã æ¬èšäºãéããŠå°ãã§ãã€ã¡ãŒãžãæŽããŠé ããŸããšå¹žãã§ãã ä»ã«ãé¢çœãããªæ©èœã¯ãããšæããŸãã®ã§ããŸããæ©äŒãããã°æçš¿ããããšæããŸãã ç§ããã¯ä»¥äžãšãªããŸããæåŸãŸã§ãèªã¿é ãããããšãããããŸããã ææ¥ã®èšäºã®æ
åœã¯ãšã³ãžãã¢ã®é«å±±ããã§ãããæ¥œãã¿ã«ã æ ªåŒäŒç€Ÿ ãšãã°ã¢ æ£ç€Ÿå¡ã®æ±äººäžèЧ hrmos.co