Default batch size now takes into account lambda concurrency

Author: Rayz0r

src/python/driver.py

@@ -103,7 +103,7 @@ def write_job_config(job_id, job_bucket, n_mappers, r_func, r_handler):
 for obj in s3.Bucket(bucket).objects.filter(Prefix=config["prefix"]).all():
     all_keys.append(obj)
 
-bsize = lambdautils.compute_batch_size(all_keys, lambda_memory)
+bsize = lambdautils.compute_batch_size(all_keys, lambda_memory, concurrent_lambdas)
 batches = lambdautils.batch_creator(all_keys, bsize)
 n_mappers = len(batches)
 document = xray_recorder.current_subsegment()

src/python/jobinfo.json

@@ -1,6 +1,6 @@
 {
-    "jobBucket": "smallya-useast-1", 
-    "mapCount": 29, 
+    "jobBucket": "aws-bigdata-mapreduce-sr18", 
+    "mapCount": 202, 
     "reducerFunction": "BL-reducer-bl-release", 
     "reducerHandler": "reducer.lambda_handler", 
     "jobId": "bl-release"

src/python/lambdautils.py

@@ -45,7 +45,7 @@ def create_lambda_function(self):
                       Description = self.function_name,
                       MemorySize = self.memory,
                       Timeout =  self.timeout,
-                      TracingConfig={'Mode':'Active'}
+                      TracingConfig={'Mode':'PassThrough'}
                     )
         self.function_arn = response['FunctionArn']
         print response
@@ -127,7 +127,7 @@ def cleanup_logs(cls, func_name):
         response = log_client.delete_log_group(logGroupName='/aws/lambda/' + func_name)
         return response
 
-def compute_batch_size(keys, lambda_memory, gzip=False):
+def compute_batch_size(keys, lambda_memory, concurrent_lambdas):
     max_mem_for_data = 0.6 * lambda_memory * 1000 * 1000; 
     size = 0.0
     for key in keys:
@@ -137,8 +137,11 @@ def compute_batch_size(keys, lambda_memory, gzip=False):
             size += key.size
     avg_object_size = size/len(keys)
     print "Dataset size: %s, nKeys: %s, avg: %s" %(size, len(keys), avg_object_size)
-    b_size = int(round(max_mem_for_data/avg_object_size))
-    return b_size 
+    if avg_object_size < max_mem_for_data and len(keys) < concurrent_lambdas:
+        b_size = 1
+    else:
+        b_size = int(round(max_mem_for_data/avg_object_size))
+    return b_size
 
 def batch_creator(all_keys, batch_size):
     '''

src/python/reducerCoordinator.py

@@ -60,7 +60,7 @@ def get_mapper_files(files):
 
 def get_reducer_batch_size(keys):
     #TODO: Paramertize memory size
-    batch_size = lambdautils.compute_batch_size(keys, 1536)
+    batch_size = lambdautils.compute_batch_size(keys, 1536, 1000)
     return max(batch_size, 2) # At least 2 in a batch - Condition for termination
 
 def check_job_done(files):