Merge pull request Azure#13104 from m-ruiz21/master

 Removed Unnecessary / Problematic to_csv from Cloud Watch -> S3 Lambda Connectors

Author: v-atulyadav

DataConnectors/AWS-S3/CloudWatchLambdaFunction.py

@@ -1,7 +1,7 @@
 
 import boto3
 import json
-import csv
+import gzip
 import time
 import pandas as pd
 from datetime import datetime
@@ -66,9 +66,13 @@ def lambda_handler(event, context):
 
             sanitized_stream_name = log_stream_name.replace('/', '_')
 
+            file_path = f'/tmp/{OUTPUT_FILE_NAME}_{sanitized_stream_name}.gz'
+
             # Export data to temporary file in the right format, which will be deleted as soon as the session ends
-            fileToS3.to_csv( f'/tmp/{OUTPUT_FILE_NAME}_{sanitized_stream_name}.gz', index=False, header=False, compression='gzip', sep = ' ', escapechar=' ',  doublequote=False, quoting=csv.QUOTE_NONE)
-            
+            text_content = '\n'.join(fileToS3['message'].astype(str).values)
+            with gzip.open(file_path, 'wt', encoding='utf-8') as f:
+                f.write(text_content)
+    
             # Upload data to desired folder in bucket
             s3.Bucket(BUCKET_NAME).upload_file(f'/tmp/{OUTPUT_FILE_NAME}_{sanitized_stream_name}.gz', f'{BUCKET_PREFIX}{OUTPUT_FILE_NAME}_{sanitized_stream_name}.gz')
 

DataConnectors/AWS-S3/CloudWatchLambdaFunction_V2.py

@@ -1,7 +1,7 @@
 
+import gzip
 import boto3
 import json
-import csv
 import time
 import pandas as pd
 from datetime import datetime
@@ -99,12 +99,14 @@ def lambda_handler(event, context):
                                 # If the column exists, drop it
                                 df.drop(columns=["ingestionTime"], inplace=True)
                                 fileToS3 = df
-                                try:                
-                                    # Export data to temporary file in the right format, which will be deleted as soon as the session ends
-                                    fileToS3.to_csv( f'/tmp/{output_File_Name}.gz', index=False, header=False, compression='gzip', sep = ' ', escapechar=' ',  doublequote=False, quoting=csv.QUOTE_NONE)
+                                try:     
+                                    file_path = f'/tmp/{output_File_Name}.gz'           
+                                    text_content = '\n'.join(fileToS3['message'].astype(str).values)
+                                    with gzip.open(file_path, 'wt', encoding='utf-8') as f:
+                                        f.write(text_content)
 
                                     # Upload data to desired folder in bucket
-                                    s3.Bucket(BUCKET_NAME).upload_file(f'/tmp/{output_File_Name}.gz', f'{BUCKET_PREFIX}{output_File_Name}.gz')
+                                    s3.Bucket(BUCKET_NAME).upload_file(file_path, f'{BUCKET_PREFIX}{output_File_Name}.gz')
                                 except Exception as e:                
                                     print("Error exporting to S3 %s %s: %s" % (key["logGroupName"], key["logStreamName"], getattr(e, 'message', repr(e))))
                             else:

DataConnectors/AWS-S3/CloudWatchPushBasedLambdaFunction.py

@@ -59,7 +59,10 @@ def lambda_handler(event, context):
         sanitized_stream_name = log_data['logStream'].replace('/', '_')
         first_timestamp = df['timestamp'].iloc[0]
         file_path = f'/tmp/{sanitized_stream_name}_{first_timestamp}.gz'
-        fileToS3.to_csv(file_path, index=False, header=False, compression='gzip', sep=' ', escapechar=' ', doublequote=False, quoting=csv.QUOTE_NONE)
+        
+        text_content = '\n'.join(fileToS3['message'].astype(str).values)
+        with gzip.open(file_path, 'wt', encoding='utf-8') as f:
+            f.write(text_content)
 
         # Update S3 path with or without OUTPUT_FILE_NAME
         s3_key = f'{BUCKET_PREFIX}{current_date}/'