simple test DAG

airflow/dags/import_skg_if.py

@@ -31,7 +31,7 @@ OPENSEARCH_PASSWD = Variable.get("OPENSEARCH_PASSWORD", "admin")
 
 ENTITIES = ["datasources", "grants", "organizations", "persons", "products", "topics", "venues"]
 
-BULK_PARALLELISM = 16
+BULK_PARALLELISM = 10
 
 #
 
@@ -114,15 +114,15 @@ def skg_if_pipeline():
                     }
 
                 },
-                # "mappings": mappings[entity]
-                "mappings":{
-                    "dynamic": False,
-                    "properties": {
-                        "local_identifier": {
-                            "type": "keyword"
-                        }
-                    }
-                }
+                "mappings": mappings[entity]
+                # "mappings":{
+                #     "dynamic": False,
+                #     "properties": {
+                #         "local_identifier": {
+                #             "type": "keyword"
+                #         }
+                #     }
+                # }
             })
 
     def compute_batches(ds=None, **kwargs):
@@ -148,7 +148,7 @@ def skg_if_pipeline():
                         name="base",
                         resources=k8s.V1ResourceRequirements(
                             requests={
-                                "cpu": "1",
+                                "cpu": "550m",
                                 "memory": "256Mi"
                             }
                         )