Merge pull request #163 from pycontw/airflow-3-bug-fixes

Add airflow triggerer, update config to fit our limited resource and fix bug found in posts_insights dag

Author: henry410213028

airflow.cfg

@@ -23,6 +23,9 @@ auth_manager = airflow.providers.fab.auth_manager.fab_auth_manager.FabAuthManage
 [dag_processor]
 # How long before timing out a DagFileProcessor, which processes a dag file
 dag_file_processor_timeout = 600
+# We don't really need it based on how we deploy dags.
+# But set it to a large enough 60 * 24 (1 day) to avoid high CPU usage
+min_file_process_interval = 1440
 
 [database]
 # The SqlAlchemy connection string to the metadata database.
@@ -34,7 +37,8 @@ external_db_managers = airflow.providers.fab.auth_manager.models.db.FABDBManager
 
 [api]
 # Number of workers to run the Gunicorn web server
-workers = 2
+# WARNING:: DO NOT increase this number. Due to our limited resources, increasing it to 2 breaks API server
+workers = 1
 
 # Number of seconds the gunicorn webserver waits before timing out on a worker
 worker_timeout = 600

dags/utils/posts_insights/base.py

@@ -126,14 +126,21 @@ def _dump_to_bigquery(
             logger.info(f"No {dump_type} to dump!")
             return
 
+        if dump_type == "posts":
+            target_table = self.POST_TABLE_NAME
+        elif dump_type == "posts insights":
+            target_table = self.INSIGHT_TABLE_NAME
+        else:
+            raise ValueError(f"Unexpected dump_type {dump_type}")
+
         job_config = bigquery.LoadJobConfig(
             schema=bq_schema_fields,
             write_disposition="WRITE_APPEND",
         )
         try:
             job = self.bq_client.load_table_from_json(
                 posts,
-                f"pycontw-225217.ods.{self.INSIGHT_TABLE_NAME}",
+                f"pycontw-225217.ods.{target_table}",
                 job_config=job_config,
             )
             job.result()

docker-compose-dev.yml

@@ -63,6 +63,21 @@ services:
       airflow-init:
         condition: service_completed_successfully
 
+  airflow-triggerer:
+    <<: *docker-common
+    container_name: airflow-triggerer
+    command: triggerer
+    healthcheck:
+      test: ["CMD-SHELL", 'airflow jobs check --job-type TriggererJob --hostname "$${HOSTNAME}"']
+      interval: 30s
+      timeout: 10s
+      retries: 5
+      start_period: 30s
+    restart: always
+    depends_on:
+      airflow-init:
+        condition: service_completed_successfully
+
   airflow-init:
     <<: *docker-common
     entrypoint: /bin/bash

docker-compose.yml

@@ -58,6 +58,21 @@ services:
       airflow-init:
         condition: service_completed_successfully
 
+  airflow-triggerer:
+    <<: *docker-common
+    container_name: airflow-triggerer
+    command: triggerer
+    healthcheck:
+      test: ["CMD-SHELL", 'airflow jobs check --job-type TriggererJob --hostname "$${HOSTNAME}"']
+      interval: 30s
+      timeout: 10s
+      retries: 5
+      start_period: 30s
+    restart: always
+    depends_on:
+      airflow-init:
+        condition: service_completed_successfully
+
   airflow-init:
     <<: *docker-common
     entrypoint: /bin/bash