Nxtra

Serverless data pipelines: ETL workflow with Step Functions and Athena

This blog is Part 3 of a multi-part series around analysing Flanders’ traffic whilst leveraging the power of cloud components!
For part 1 see: https://medium.com/cloudway/real-time-data-processing-with-kinesis-data-analytics-ad52ad338c6d
For part 2 see: https://medium.com/cloubis/serverless-data-transform-with-kinesis-e468abd33409

What is our goal?

This blog aims to explore the use of the AWS Glue service in conjunction with the AWS Athena service to repartition raw streaming data events.
We previously landed these events on an Amazon S3 bucket partitioned according to the processing time on Kinesis.

Nxtra

    "AccessDynamoDBApiResourcesPolicy": {
      "DependsOn": ["LambdaExecutionRole"],
      "Type": "AWS::IAM::Policy",
      "Properties": {
        "PolicyName": "amplify-lambda-execution-policy",
        "Roles": [
          {
            "Ref": "LambdaExecutionRole"
          }
        ],

Nxtra

const AWS = require("aws-sdk");

const docClient = new AWS.DynamoDB.DocumentClient();

const POSTTABLE = process.env.POSTTABLE;
const COMMENTTABLE = process.env.COMMENTTABLE;

const resolvers = {
  Mutation: {
    deletePostAndComments: (event) => {

Nxtra

type Blog @model {
  id: ID!
  name: String!
  posts: [Post] @connection(keyName: "byBlog", fields: ["id"])
}

type Post @model @key(name: "byBlog", fields: ["blogID"]) {
  id: ID!
  title: String!
  blogID: ID!

Nxtra

AWSTemplateFormatVersion: "2010-09-09"
Description: Billing Alerts for your AWS Account

Parameters:
  Email:
    Type: String
    Default: youremail@somerandomdomain.com
    Description: The email address to receive alerts per email
  Phone:

Nxtra

describe("Example test", () => {
  before(() => {
    cy.signIn();
  });

  after(() => {
    cy.clearLocalStorageSnapshot();
    cy.clearLocalStorage();
  });

Nxtra

Cypress.Commands.add("signIn", () => {
  cy.then(() => Auth.signIn(username, password)).then((cognitoUser) => {
    const idToken = cognitoUser.signInUserSession.idToken.jwtToken;
    const accessToken = cognitoUser.signInUserSession.accessToken.jwtToken;

    const makeKey = (name) => `CognitoIdentityServiceProvider.${cognitoUser.pool.clientId}.${cognitoUser.username}.${name}`;

    cy.setLocalStorage(makeKey("accessToken"), accessToken);
    cy.setLocalStorage(makeKey("idToken"), idToken);
    cy.setLocalStorage(

Nxtra

const Auth = require("aws-amplify").Auth;

import "cypress-localstorage-commands";

const username = Cypress.env("username");
const password = Cypress.env("password");
const userPoolId = Cypress.env("userPoolId");
const clientId = Cypress.env("clientId");

const awsconfig = {

Nxtra

CREATE OR REPLACE PUMP "YOUR_IN_APPLICATION_STREAM" AS
    INSERT INTO "YOUR_IN_APPLICATION_PUMP" ("uniqueId", "currentSpeed", ..., "location")
        SELECT STREAM
        "sdi"."uniqueId",
        "sdi"."currentSpeed",
        ...,
        "ml"."locatie",
        FROM "SPEED_DIFF_INDICATOR_SQL_STREAM" AS "sdi" LEFT JOIN "measurementLocations" as "ml"
        ON "sdi"."uniqueId" = "ml"."id";

Nxtra

CREATE OR REPLACE PUMP "SPEED_CHANGE_PUMP" AS
    INSERT INTO "SPEED_CHANGE_SQL_STREAM"
        SELECT STREAM "s"."uniqueId",
            LAG("s"."speed", 1, "s"."speed") OVER CURRENT_WINDOW AS "previousSpeed",
            "s"."speed" AS "currentSpeed"
        FROM "SPEED_SQL_STREAM" AS "s"
        WINDOW CURRENT_WINDOW AS (PARTITION BY "s"."uniqueId" ROWS 3 PRECEDING);