Uses Athena's INSERT INTO, upgrades to node 12

Author: steffeng

README.md

@@ -19,7 +19,7 @@ The application has two main parts:
 
     ![infrastructure-overview](images/moveAccessLogs.png)
 
-- An hourly scheduled AWS Lambda function `transformPartition` that runs a [Create Table As Select](https://docs.aws.amazon.com/athena/latest/ug/ctas.html) (CTAS) query on a single partition per run, taking one hour of data into account. It writes the content of the partition to the Apache Parquet format into the `<StackName>-cf-access-logs` S3 bucket.
+- An hourly scheduled AWS Lambda function `transformPartition` that runs an [INSERT INTO](https://docs.aws.amazon.com/athena/latest/ug/insert-into.html) query on a single partition per run, taking one hour of data into account. It writes the content of the partition to the Apache Parquet format into the `<StackName>-cf-access-logs` S3 bucket.
 
     ![infrastructure-overview](images/transformPartition.png)
 

functions/transformPartition.js

@@ -7,9 +7,6 @@ const sourceTable = process.env.SOURCE_TABLE;
 const targetTable = process.env.TARGET_TABLE;
 const database = process.env.DATABASE;
 
-// s3 URL to write CTAS results to (including trailing slash)
-const athenaCtasResultsLocation = process.env.ATHENA_CTAS_RESULTS_LOCATION;
-
 // get the partition of 2hours ago
 exports.handler = async (event, context, callback) => {
   var partitionHour = new Date(Date.now() - 120 * 60 * 1000);
@@ -20,35 +17,14 @@ exports.handler = async (event, context, callback) => {
 
   console.log('Transforming Partition', { year, month, day, hour });
 
-  var intermediateTable = `ctas_${year}_${month}_${day}_${hour}`;
-
   var ctasStatement = `
-    CREATE TABLE ${database}.${intermediateTable}
-    WITH ( format='PARQUET',
-        external_location='${athenaCtasResultsLocation}year=${year}/month=${month}/day=${day}/hour=${hour}',
-        parquet_compression = 'SNAPPY')
-    AS SELECT *
+    INSERT INTO ${database}.${targetTable}
+    SELECT *
     FROM ${database}.${sourceTable}
     WHERE year = '${year}'
         AND month = '${month}'
         AND day = '${day}'
         AND hour = '${hour}';`;
 
-  var dropTableStatement = `DROP TABLE ${database}.${intermediateTable};`;
-
-  var createNewPartitionStatement = `
-    ALTER TABLE ${database}.${targetTable}
-    ADD IF NOT EXISTS 
-    PARTITION (
-        year = '${year}',
-        month = '${month}',
-        day = '${day}',
-        hour = '${hour}' );`;
-
-  await util.runQuery(ctasStatement).then(
-    () => Promise.all([
-      util.runQuery(dropTableStatement),
-      util.runQuery(createNewPartitionStatement)
-    ])
-  );
+  await util.runQuery(ctasStatement);
 }

images/transformPartition.png

@@ -45,7 +45,7 @@ Resources:
         Properties:
             CodeUri: functions/
             Handler: transformPartition.handler
-            Runtime: nodejs8.10
+            Runtime: nodejs12.x
             Timeout: 900
             Policies:
             - Version: '2012-10-17'
@@ -83,7 +83,6 @@ Resources:
                     TARGET_TABLE: !Ref PartitionedParquetTable
                     DATABASE: !Ref CfLogsDatabase
                     ATHENA_QUERY_RESULTS_LOCATION: !Sub "s3://${ResourcePrefix}-${AWS::AccountId}-cf-access-logs/athena-query-results"
-                    ATHENA_CTAS_RESULTS_LOCATION: !Sub "s3://${ResourcePrefix}-${AWS::AccountId}-cf-access-logs/${ParquetKeyPrefix}"
             Events:
               HourlyEvt:
                 Type: Schedule
@@ -94,7 +93,7 @@ Resources:
         Properties:
             CodeUri: functions/
             Handler: createPartitions.handler
-            Runtime: nodejs8.10
+            Runtime: nodejs12.x
             Timeout: 5
             Policies:
             - Version: '2012-10-17'
@@ -135,7 +134,7 @@ Resources:
         Properties:
             CodeUri: functions/
             Handler: moveAccessLogs.handler
-            Runtime: nodejs8.10
+            Runtime: nodejs12.x
             Timeout: 30
             Policies:
             - Version: '2012-10-17'