Viz EC2 Services to Lambda - Part 3b - Wrap Up

Core/LAMBDA/layers/main.tf

@@ -332,5 +332,5 @@ output "yaml" {
 }
 
 output "dask" {
-  value = resource.aws_lambda_layer_version.yaml
+  value = resource.aws_lambda_layer_version.dask
 }

Core/LAMBDA/viz_functions/main.tf

@@ -394,12 +394,12 @@ resource "aws_s3_object" "python_preprocessing_zip_upload" {
 }
 
 #########################
-#### 2GB RAM Version ####
+#### 3GB RAM Version ####
 #########################
-resource "aws_lambda_function" "viz_python_preprocessing_2GB" {
-  function_name = "hv-vpp-${var.environment}-viz-python-preprocessing"
+resource "aws_lambda_function" "viz_python_preprocessing_3GB" {
+  function_name = "hv-vpp-${var.environment}-viz-python-preprocessing-3GB"
   description   = "Lambda function to create max streamflow files for NWM data"
-  memory_size   = 2048
+  memory_size   = 3072
   ephemeral_storage {
     size = 6656
   }
@@ -440,15 +440,15 @@ resource "aws_lambda_function" "viz_python_preprocessing_2GB" {
   ]
 
   tags = {
-    "Name" = "hv-vpp-${var.environment}-viz-python-preprocessing-2GB"
+    "Name" = "hv-vpp-${var.environment}-viz_python_preprocessing_3GB"
   }
 }
 
 #########################
 #### 10GB RAM Version ####
 #########################
 resource "aws_lambda_function" "viz_python_preprocessing_10GB" {
-  function_name = "hv-vpp-${var.environment}-viz-python-preprocessing"
+  function_name = "hv-vpp-${var.environment}-viz-python-preprocessing-10GB"
   description   = "Lambda function to create max streamflow files for NWM data"
   memory_size   = 10240
   ephemeral_storage {
@@ -491,7 +491,7 @@ resource "aws_lambda_function" "viz_python_preprocessing_10GB" {
   ]
 
   tags = {
-    "Name" = "hv-vpp-${var.environment}-viz-python-preprocessing-2GB"
+    "Name" = "hv-vpp-${var.environment}-viz_python_preprocessing_10GB"
   }
 }
 
@@ -1000,8 +1000,8 @@ module "image-based-lambdas" {
 ########################################################################################################################################
 ########################################################################################################################################
 
-output "python_preprocessing_2GB" {
-  value = aws_lambda_function.viz_python_preprocessing_2GB
+output "python_preprocessing_3GB" {
+  value = aws_lambda_function.viz_python_preprocessing_3GB
 }
 
 output "python_preprocessing_10GB" {

Core/LAMBDA/viz_functions/viz_initialize_pipeline/lambda_function.py

@@ -499,6 +499,7 @@ def generate_python_preprocessing_file_list(self, file_groups):
         for file_group in file_groups:
             product = file_group['product'] 
             config = file_group['config'] if file_group.get('config') else None
+            lambda_ram = file_group['lambda_ram'] if file_group.get('lambda_ram') else None
             output_file = file_group['output_file']
 
             token_dict = get_file_tokens(output_file)
@@ -510,6 +511,7 @@ def generate_python_preprocessing_file_list(self, file_groups):
                 "fileset_bucket": python_preprocesing_file_set[0]['bucket'],
                 "product": product,
                 "config": config,
+                "lambda_ram": lambda_ram,
                 "output_file": formatted_output_file,
                 "output_file_bucket": os.environ['PYTHON_PREPROCESSING_BUCKET'],
             })

Core/LAMBDA/viz_functions/viz_initialize_pipeline/product_configs/analysis_assim/ana_anomaly.yml

@@ -10,6 +10,7 @@ python_preprocessing:
     file_step: 1H
     file_window: P7D
     product: anomaly
+    lambda_ram: 10gb
     config: 7
     output_file: viz_ingest/analysis_assim/{{datetime:%Y%m%d}}/{{datetime:%H}}_ana_7day_anomaly.csv
     target_table: ingest.ana_7day_anomaly
@@ -18,6 +19,7 @@ python_preprocessing:
     file_step: 1H
     file_window: P14D
     product: anomaly
+    lambda_ram: 10gb
     config: 14
     output_file: viz_ingest/analysis_assim/{{datetime:%Y%m%d}}/{{datetime:%H}}_ana_14day_anomaly.csv
     target_table: ingest.ana_14day_anomaly

Core/LAMBDA/viz_functions/viz_initialize_pipeline/product_configs/analysis_assim/ana_past_14day_max_high_flow_magnitude.yml

@@ -10,13 +10,15 @@ python_preprocessing:
     file_step: 1H
     file_window: P7D
     product: max_values
+    lambda_ram: 3gb
     output_file: max_flows/analysis_assim/{{datetime:%Y%m%d}}/ana_7day_00_max_flows.nc
     target_table: ingest.nwm_channel_rt_ana_7day_max
     target_keys: (feature_id, streamflow)
   - file_format: common/data/model/com/nwm/{{variable:NWM_DATAFLOW_VERSION}}/nwm.{{datetime:%Y%m%d}}/analysis_assim/nwm.t{{datetime:%H}}z.analysis_assim.channel_rt.tm00.conus.nc
     file_step: 1H
     file_window: P14D
     product: max_values
+    lambda_ram: 3gb
     output_file: max_flows/analysis_assim/{{datetime:%Y%m%d}}/ana_14day_00_max_flows.nc
     target_table: ingest.nwm_channel_rt_ana_14day_max
     target_keys: (feature_id, streamflow)

Core/LAMBDA/viz_functions/viz_initialize_pipeline/product_configs/analysis_assim/ana_past_14day_max_inundation.yml

@@ -10,13 +10,15 @@ python_preprocessing:
     file_step: 1H
     file_window: P7D
     product: max_values
+    lambda_ram: 3gb
     output_file: max_flows/analysis_assim/{{datetime:%Y%m%d}}/ana_7day_00_max_flows.nc
     target_table: ingest.nwm_channel_rt_ana_7day_max
     target_keys: (feature_id, streamflow)
   - file_format: common/data/model/com/nwm/{{variable:NWM_DATAFLOW_VERSION}}/nwm.{{datetime:%Y%m%d}}/analysis_assim/nwm.t{{datetime:%H}}z.analysis_assim.channel_rt.tm00.conus.nc
     file_step: 1H
     file_window: P14D
     product: max_values
+    lambda_ram: 3gb
     output_file: max_flows/analysis_assim/{{datetime:%Y%m%d}}/ana_14day_00_max_flows.nc
     target_table: ingest.nwm_channel_rt_ana_14day_max
     target_keys: (feature_id, streamflow)

Core/LAMBDA/viz_functions/viz_initialize_pipeline/product_configs/medium_range_mem1/mrf_gfs_5day_max_high_water_probability.yml

@@ -8,6 +8,7 @@ python_preprocessing:
     file_step: None
     file_window: None
     product: high_water_probability
+    lambda_ram: 3gb
     output_file: viz_ingest/medium_range_mem1/{{datetime:%Y%m%d}}/{{datetime:%H}}_mrf_gfs_5day_max_high_water_probability.csv
     target_table: ingest.mrf_gfs_5day_max_high_water_prob
     target_keys: (feature_id)

Core/LAMBDA/viz_functions/viz_initialize_pipeline/product_configs/medium_range_mem1/mrf_gfs_5day_rapid_onset_flooding_probability.yml

@@ -8,6 +8,7 @@ python_preprocessing:
     file_step: None
     file_window: None
     product: rapid_onset_flooding_probability
+    lambda_ram: 10gb
     output_file: viz_ingest/medium_range_mem1/{{datetime:%Y%m%d}}/{{datetime:%H}}_mrf_gfs_5day_rapid_onset_flooding_probability.csv
     target_table: ingest.mrf_gfs_5day_rof_prob
     target_keys: (feature_id)

Core/LAMBDA/viz_functions/viz_initialize_pipeline/product_configs/short_range/srf_12hr_max_high_water_probability.yml

@@ -8,6 +8,7 @@ python_preprocessing:
     file_step: 1H
     file_window: PT7H
     product: high_water_probability
+    lambda_ram: 3gb
     output_file: viz_ingest/short_range/{{datetime:%Y%m%d}}/{{datetime:%H}}_srf_12hr_max_high_water_probability.csv
     target_table: ingest.srf_12hr_max_high_water_prob
     target_keys: (feature_id)

Core/LAMBDA/viz_functions/viz_initialize_pipeline/product_configs/short_range/srf_12hr_rapid_onset_flooding_probability.yml

@@ -8,6 +8,7 @@ python_preprocessing:
     file_step: 1H
     file_window: PT7H
     product: rapid_onset_flooding_probability
+    lambda_ram: 3gb
     output_file: viz_ingest/short_range/{{datetime:%Y%m%d}}/{{datetime:%H}}_srf_12hr_rapid_onset_flooding_probability.csv
     target_table: ingest.srf_12hr_rof_prob
     target_keys: (feature_id)

Core/StepFunctions/main.tf

@@ -10,7 +10,7 @@ variable "environment" {
   type        = string
 }
 
-variable "python_preprocessing_2GB_arn" {
+variable "python_preprocessing_3GB_arn" {
   type        = string
 }
 
@@ -152,7 +152,7 @@ resource "aws_sfn_state_machine" "viz_pipeline_step_function" {
     role_arn = var.viz_lambda_role
 
     definition = templatefile("${path.module}/viz_processing_pipeline.json.tftpl", {
-        python_preprocessing_2GB_arn = var.python_preprocessing_2GB_arn
+        python_preprocessing_3GB_arn = var.python_preprocessing_3GB_arn
         python_preprocessing_10GB_arn = var.python_preprocessing_10GB_arn
         db_postprocess_sql_arn = var.db_postprocess_sql_arn
         db_ingest_arn      = var.db_ingest_arn

Core/StepFunctions/viz_processing_pipeline.json.tftpl

@@ -28,20 +28,66 @@
       "Resource": "arn:aws:states:::aws-sdk:cloudwatchlogs:putLogEvents",
       "ResultPath": null
     },
-    "Python Preprocessing": {
+     "Python Preprocessing": {
       "Type": "Map",
       "ItemProcessor": {
         "ProcessorConfig": {
           "Mode": "INLINE"
         },
-        "StartAt": "Python Preprocess",
+        "StartAt": "Small or Large Lambda Function",
         "States": {
-          "Python Preprocess": {
+          "Small or Large Lambda Function": {
+            "Type": "Choice",
+            "Choices": [
+              {
+                "Variable": "$.python_preprocessing.lambda_ram",
+                "StringEquals": "10gb",
+                "Next": "Python Preprocess - 10GB"
+              }
+            ],
+            "Default": "Python Preprocess - 3GB"
+          },
+          "Python Preprocess - 3GB": {
+            "Type": "Task",
+            "Resource": "arn:aws:states:::lambda:invoke",
+            "OutputPath": "$.Payload",
+            "Parameters": {
+              "FunctionName": "${python_preprocessing_3GB_arn}",
+              "Payload": {
+                "args.$": "$",
+                "step": "python_preprocessing"
+              }
+            },
+            "Retry": [
+              {
+                "ErrorEquals": [
+                  "Lambda.ServiceException",
+                  "Lambda.AWSLambdaException",
+                  "Lambda.SdkClientException",
+                  "Lambda.TooManyRequestsException"
+                ],
+                "IntervalSeconds": 2,
+                "MaxAttempts": 6,
+                "BackoffRate": 2
+              },
+              {
+                "ErrorEquals": [
+                  "MissingS3FileException"
+                ],
+                "BackoffRate": 1,
+                "IntervalSeconds": 120,
+                "MaxAttempts": 35,
+                "Comment": "Missing S3 File"
+              }
+            ],
+            "End": true
+          },
+          "Python Preprocess - 10GB": {
             "Type": "Task",
             "Resource": "arn:aws:states:::lambda:invoke",
             "OutputPath": "$.Payload",
             "Parameters": {
-              "FunctionName": "${python_preprocessing_2GB_arn}",
+              "FunctionName": "${python_preprocessing_10GB_arn}",
               "Payload": {
                 "args.$": "$",
                 "step": "python_preprocessing"
@@ -683,7 +729,7 @@
                   "Type": "Task",
                   "Resource": "arn:aws:states:::lambda:invoke",
                   "Parameters": {
-                    "FunctionName": "${python_preprocessing_2GB_arn}",
+                    "FunctionName": "${python_preprocessing_3GB_arn}",
                     "Payload": {
                       "args.$": "$",
                       "step": "fim_config_max_file"

Core/main.tf

@@ -635,7 +635,7 @@ module "step-functions" {
   db_ingest_arn                     = module.viz-lambda-functions.db_ingest.arn
   raster_processing_arn             = module.viz-lambda-functions.raster_processing.arn
   publish_service_arn               = module.viz-lambda-functions.publish_service.arn
-  python_preprocessing_2GB_arn      = module.viz-lambda-functions.python_preprocessing_2GB.arn
+  python_preprocessing_3GB_arn      = module.viz-lambda-functions.python_preprocessing_3GB.arn
   python_preprocessing_10GB_arn     = module.viz-lambda-functions.python_preprocessing_10GB.arn
   hand_fim_processing_arn           = module.viz-lambda-functions.hand_fim_processing.arn
   schism_fim_processing_arn         = module.viz-lambda-functions.schism_fim_processing.arn