Add dataset mapping and reference it in s3_to_json_s3 script

.github/workflows/upload-and-deploy.yaml

@@ -55,8 +55,8 @@ jobs:
       - name: Copy files to templates bucket
         run: |
           # copy glue python scripts to bucket
-          aws s3 sync src/glue/jobs/ \
-            s3://$CFN_BUCKET/$REPO_NAME/$GITHUB_REF_NAME/glue/jobs/
+          aws s3 sync src/glue/ \
+            s3://$CFN_BUCKET/$REPO_NAME/$GITHUB_REF_NAME/glue/
 
           # copy CFN templates to bucket
           aws s3 sync templates/ \

src/glue/jobs/s3_to_json_s3.py

@@ -8,6 +8,7 @@
 import sys
 import zipfile
 from datetime import datetime
+from urllib.parse import urlparse
 
 import boto3
 from awsglue.utils import getResolvedOptions
@@ -23,18 +24,40 @@
 args = getResolvedOptions(
         sys.argv,
         ["WORKFLOW_NAME",
-         "WORKFLOW_RUN_ID"])
+         "WORKFLOW_RUN_ID",
+         "scriptLocation"])
 workflow_run_properties = glue_client.get_workflow_run_properties(
         Name=args["WORKFLOW_NAME"],
         RunId=args["WORKFLOW_RUN_ID"])["RunProperties"]
 
-def process_record(s3_obj, s3_obj_metadata):
+def get_dataset_mapping(script_location):
+    script_location = urlparse(script_location)
+    dataset_mapping_bucket = script_location.netloc
+    dataset_mapping_key = "BridgeDownstream/main/glue/resources/dataset_mapping.json"
+    dataset_mapping_fname = os.path.basename(dataset_mapping_key)
+    dataset_mapping_file = s3_client.download_file(
+            Bucket=dataset_mapping_bucket,
+            Key=dataset_mapping_key,
+            Filename=dataset_mapping_fname)
+    with open(dataset_mapping_fname, "r") as f:
+        dataset_mapping = json.load(f)
+    return(dataset_mapping)
+
+
+def process_record(s3_obj, s3_obj_metadata, dataset_mapping):
     created_on = datetime.fromtimestamp(
             int(s3_obj_metadata["createdon"]) / 1000)
+    this_dataset_mapping = dataset_mapping[
+            "appVersion"][s3_obj_metadata["appversion"]]["dataset"]
     with zipfile.ZipFile(io.BytesIO(s3_obj["Body"].read())) as z:
         contents = z.namelist()
         for json_path in z.namelist():
-            dataset_name = os.path.splitext(json_path)[0]
+            dataset_key = os.path.splitext(json_path)[0]
+            dataset_version = this_dataset_mapping[dataset_key]
+            if dataset_version == "v1":
+                dataset_name = dataset_key
+            else:
+                dataset_name = f"{dataset_key}_{dataset_version}"
             os.makedirs(dataset_name, exist_ok=True)
             with z.open(json_path, "r") as p:
                 j = json.load(p)
@@ -86,9 +109,12 @@ def process_record(s3_obj, s3_obj_metadata):
                             Metadata = s3_obj_metadata)
 
 logger.info(f'Retrieving S3 object for Bucket {workflow_run_properties["source_bucket"]} and Key {workflow_run_properties["source_key"]}')
+dataset_mapping = get_dataset_mapping(
+        script_location=args["scriptLocation"])
 s3_obj = s3_client.get_object(
         Bucket = workflow_run_properties["source_bucket"],
         Key = workflow_run_properties["source_key"])
 process_record(
         s3_obj = s3_obj,
-        s3_obj_metadata=s3_obj["Metadata"])
+        s3_obj_metadata=s3_obj["Metadata"],
+        dataset_mapping=dataset_mapping)

src/glue/resources/dataset_mapping.json

@@ -0,0 +1,19 @@
+{
+  "appVersion":
+    {
+      "version 1.0.2, build 57":
+        {
+          "dataset":
+            {
+              "answers": "v1",
+              "info": "v2",
+              "metadata": "v1",
+              "microphone_levels": "v1",
+              "motion": "v1",
+              "taskData": "v1",
+              "taskResult": "v1",
+              "weather": "v1"
+            }
+        }
+    }
+}