fix

airflow/dags/batch-data-import.py

@@ -29,7 +29,7 @@ def generate_job(name : str, command : str, instance_size : str = "small", taskg
     node_selector = None
     container_resources = None
     if instance_size == "medium" or instance_size == "large":
-        node_selector = {"cloud.google.com/gke-nodepool": "batch-burst-node-pool"}
+        node_selector = {"cloud.google.com/gke-nodepool": "large-batch-burst"}
         tolerations.append(k8s.V1Toleration(effect="NoSchedule", key="BATCH_BURST", operator="Equal", value="true"))
 
         if instance_size == "medium":
@@ -39,6 +39,12 @@ def generate_job(name : str, command : str, instance_size : str = "small", taskg
 
         container_resources = k8s.V1ResourceRequirements(requests={"memory": memory_requests})
 
+    if instance_size == "index_small":
+        node_selector = {"cloud.google.com/gke-nodepool": "small-batch-burst"}
+        tolerations.append(k8s.V1Toleration(effect="NoSchedule", key="SMALL_BATCH_BURST", operator="Equal", value="true"))
+
+        container_resources = k8s.V1ResourceRequirements(requests={"memory": "4Gi"})
+
     k = KubernetesPodOperator(
       namespace='default',
       image='ghcr.io/travigo/travigo:main',
@@ -142,7 +148,7 @@ def generate_job(name : str, command : str, instance_size : str = "small", taskg
     end = DummyOperator(task_id="end")
 
     stop_linker = generate_job("stop-linker", [ "data-linker", "run", "--type", "stops" ])
-    stop_indexer = generate_job("stop-indexer", [ "indexer", "stops" ])
+    stop_indexer = generate_job("stop-indexer", [ "indexer", "stops" ], instance_size="index_small")
 
     taskgroups = {
         "small": TaskGroup("small"),

data/datasources/gb-dft.yaml

@@ -4,12 +4,12 @@ provider:
   name: Department for Transport
   website: "https://www.gov.uk/government/organisations/department-for-transport"
 datasets:
-- identifier: naptan
-  format: gb-naptan
-  source: "https://naptan.api.dft.gov.uk/v1/access-nodes?dataFormat=xml"
-  supportedobjects:
-    stops:      true
-    stopgroups: true
+# - identifier: naptan
+#   format: gb-naptan
+#   source: "https://naptan.api.dft.gov.uk/v1/access-nodes?dataFormat=xml"
+#   supportedobjects:
+#     stops:      true
+#     stopgroups: true
 - identifier: bods-gtfs-schedule
   format: gtfs-schedule
   source: "https://data.bus-data.dft.gov.uk/timetable/download/gtfs-file/all/"

pkg/api/routes/stops.go

@@ -286,6 +286,9 @@ func searchStops(c *fiber.Ctx) error {
 				"filter": queryFilters,
 			},
 		},
+		"collapse": map[string]interface{}{
+			"field": "PrimaryIdentifier.keyword",
+		},
 	}
 
 	json.NewEncoder(&queryBytes).Encode(searchQuery)
@@ -298,7 +301,8 @@ func searchStops(c *fiber.Ctx) error {
 	)
 
 	if err != nil {
-		log.Fatal().Err(err).Msg("Failed to query index")
+		log.Error().Err(err).Msg("Failed to query index")
+		return nil
 	}
 
 	responseBytes, _ := io.ReadAll(res.Body)