In this first chapter, you will learn about the differences between batch and stream processing, create your first stream, manage its permissions, write to it and read from it.

What is streaming and why does it matter?

Batch vs stream

Producers and destinations

Managing Firehose delivery streams

Getting ready for the first stream

Creating roles

IAM users vs. roles

Permissions practice

S3 bucket creation

Working with the Firehose delivery stream

Create your first Firehose stream

Writing to a Firehose stream

Reading Firehose data

Streaming in the cloud

The next step in your streaming data journey is learning how to use transformational lambda functions to go serverless. Through hands-on exercises, you will add lambda layers and trigger lambda functions on specific conditions.

Passare al serverless

Creare ed eseguire funzioni Lambda

Opzioni

Ideale per un server

Ideale per il serverless

Serverless vs server

Trigger

Handler

Layer

Componenti di una funzione Lambda

# EDIT ME!
# recordReaderS3/lambda_function.py
# This is lambda handler code that will execute each time an object is added to S3.
import json

def record_created_handler(____, ____):
    # Create a list to store new object keys
    written_objects = []

    # Iterate over each S3 event record
    for record in event['Records']:
       
        # Get the variables to check for
        event_name = record['eventName']
        bucket_name = record['s3']['bucket']['name']
        obj_key = record['s3']['object']['key']

        # Verify that the object was written to S3
        if event_name == 'ObjectCreated:____' and bucket_name == 'sd-vehicle-data':
           
            # If so, then add it to the written_objects list
            written_objects.append(obj_key)
    
    # Return the collected written objects
    return {
        "bucket": 'sd-vehicle-data',
        "new_objects": written_objects
    }

# PLEASE DO NOT EDIT
# run_lambda.py
# This executes a lambda function with a sample event.
from dcs.util.aws_lambda import * 
import json


# Clean up any old lambda functions
clean_lambdas()

# Create a new lambda 
create_lambda("recordReaderS3", "record_created_handler")

# Read the test event JSON file as dictionary
with open("./dcs/sample_events/s3_event_c2l1.json") as json_file:
    payload = json.load(json_file)

# Invoke lambda with test event, print result
result = invoke_function_and_get_message("recordReaderS3", payload)
print(json.dumps(json.loads(result), indent=4))

# EDIT ME!
# recordReaderS3/lambda_function.py
# This is lambda handler code that will execute each time an object is added to S3
import json

def record_created_handler(event, context):
    # Create a list to store new object keys
    written_objects = []

    # Iterate over each S3 event record
    for record in event['Records']:
       
        # Get the variables to check for
        event_name = record['eventName']
        bucket_name = record['s3']['bucket']['name']
        obj_key = record['s3']['object']['key']

        # Verify that the object was written to S3
        if event_name == 'ObjectCreated:Put' and bucket_name == 'sd-vehicle-data':
           
            # If so, then add it to the written_objects list
            written_objects.append(obj_key)
    
    # Return the collected written objects
    return {
        "bucket": 'sd-vehicle-data',
        "new_objects": written_objects
    }                 

# PLEASE DO NOT EDIT
# run_lambda.py
# This executes a lambda function with a sample event
from dcs.util.aws_lambda import * 
import json


# Clean up any old lambda functions
clean_lambdas()

# Create a new lambda 
create_lambda("recordReaderS3", "record_created_handler")

# Read the test event JSON file as dictionary
with open("./dcs/sample_events/s3_event_c2l1.json") as json_file:
    payload = json.load(json_file)

# Invoke lambda with test event, print result
result = invoke_function_and_get_message("recordReaderS3", payload)
print(json.dumps(json.loads(result), indent=4))

Lettura dei dati da S3 in base a un evento

La tua prima lambda live!

Aggiungere un layer Lambda

Flusso Lambda

Uso delle variabili d'ambiente

Flusso di lavoro dati serverless

Lambda a esecuzione rapida

Lambda che impiegano più tempo

Lambda veloci vs lente

Creare una Lambda attivata dal tempo

Una Lambda attivata a orario

Codice Lambda attivato a tempo

API serverless

Crea una Lambda attivata da API

Crea una Lambda per API

import boto3

lambda_client = boto3.client('lambda', aws_access_key_id="12345",
    aws_secret_access_key="12345",
    region_name='us-east-1',
    endpoint_url="http://localhost:4574")

def trigger_full_report_builder(event, context):
    async = event['____']['async']
    if async == 'true':
        exec_type = '____'
    else:
        exec_type = 'RequestResponse'
        
    result = lambda_client.____(
      FunctionName='fullReportBuilder',
      InvocationType=exec_type)
    
    return {
        'statusCode': 200,
        'headers': {
            "content-type" : "application/json"
        },
        'body': "Successfully Invoked"
    }

import awswrangler_c as wr
import boto3
# Initialize boto3 session
session = boto3.Session(aws_access_key_id = "12345", 
  aws_secret_access_key = "12345", 
  region_name="us-east-1")

def full_report_builder(event, context):
  # Read all records in the speeders folder
  speeders_total = wr.s3.read_csv('s3://sd-vehicle-data/speeders', 
                                  boto3_session = session, 
                                  delimiter=" ")

  # Write aggregated speeders file
  return wr.s3.to_csv(df = speeders_total, 
               boto3_session=session, 
               path="s3://sd-vehicle-data/speeders-full/full.csv")
               

# PLEASE DO NOT EDIT
# aws_wrangler_c.py
def read_csv(path, boto3_session, delimiter):
  bucket = path.split("//")[1].split("/")[0]
  prefix = path.split("//")[1].replace(f"{bucket}/", "")
  objects = s3.list_objects_v2(Bucket=bucket, Prefix=prefix)['Contents']
  dfs = []
  for o in objects:
      key = o['Key']
      data = s3.get_object(Bucket='sd-vehicle-data', Key=key)
      dfs.append(pd.read_csv(data['Body'], delimiter=delimiter))
  return pd.concat(dfs)
  
def to_csv(df, path, boto3_session, separator=","):
  bucket = path.split("//")[1].split("/")[0]
  key = path.split("//")[1].replace(f"{bucket}/", "")
  s3.put_object(Bucket=bucket, Key=key, Body=df.to_csv(sep=separator, index=False))
  return True

s3 = {
    "to_csv": to_csv,
    "read_csv": read_csv
}
                                                  

# PLEASE DO NOT EDIT
# run_lambda.py
# This executes a lambda function with a sample event.
from dcs.util.aws_lambda import * 
import json

# Clean up any old lambda functions
clean_lambdas()

# Create a new lambda 
create_lambda("fullReportBuilder", "full_report_builder")
create_lambda("triggerFullReportBuilder", "trigger_full_report_builder")


# Invoke lambda with test event, print result.
result = invoke_function_and_get_message("triggerFullReportBuilder", {
  "queryStringParameters": {
    "async": "true"
  },
  "isBase64Encoded": "false"
})
print(result)

import boto3

lambda_client = boto3.client('lambda', aws_access_key_id="12345",
    aws_secret_access_key="12345",
    region_name='us-east-1',
    endpoint_url="http://localhost:4574")

def trigger_full_report_builder(event, context):
    async = event['queryStringParameters']['async']
    if async == 'true':
        exec_type = 'Event'
    else:
        exec_type = 'RequestResponse'
        
    result = lambda_client.invoke(
      FunctionName='fullReportBuilder',
      InvocationType=exec_type)
    
    return {
        'statusCode': 200,
        'headers': {
            "content-type" : "application/json"
        },
        'body': "Successfully Invoked"
    }

    

import awswrangler_c as wr
import boto3
# Initialize boto3 session
session = boto3.Session(aws_access_key_id = "12345", 
  aws_secret_access_key = "12345", 
  region_name="us-east-1")

def full_report_builder(event, context):
  # Read all records in the speeders folder
  speeders_total = wr.s3.read_csv('s3://sd-vehicle-data/speeders', 
                                  boto3_session = session, 
                                  delimiter=" ")

  # Write aggregated speeders file
  return wr.s3.to_csv(df = speeders_total, 
               boto3_session=session, 
               path="s3://sd-vehicle-data/speeders-full/full.csv")
               
  

# PLEASE DO NOT EDIT
# run_lambda.py
# This executes a lambda function with a sample event.
from dcs.util.aws_lambda import * 
import json

# Clean up any old lambda functions
clean_lambdas()

# Create a new lambda 
create_lambda("fullReportBuilder", "full_report_builder")
create_lambda("triggerFullReportBuilder", "trigger_full_report_builder")

# Invoke lambda with test event, print result.
result = invoke_function_and_get_message("triggerFullReportBuilder", {
  "queryStringParameters": {
    "async": "true"
  },
  "isBase64Encoded": "false"
})
print(result)

Un'API serverless

Going serverless

You're now ready to encode and decode streaming data and analyze data directly in the stream. You will even use multiple streams to get daily vehicle top speeds.

A transformational Lambda

Transforming data inside a stream

Encoding and decoding base64

Create a transformational lambda

A barebones transformational lambda

Analyzing data in the stream

Creating a Kinesis data analytics application

Kinesis data analytics vs transformational Lambda

Building a Kinesis data analytics application

Kinesis data analytics SQL components

Using multiple streams

Delivering data from Kinesis Analytics

Get the daily top speed

Using Kinesis data analytics for alerts

Analyzing streaming data

In this final chapter, you will discover how to monitor your stream's performance using logs, metrics, alarms and dashboards. You will use Elasticsearch and build your own Kibana dashboard.

Streaming data case study

Creating an Elasticsearch cluster

RedShift vs Elasticsearch

Monitoring performance

Cloudwatch dasbhoards and alarms

Cloudwatch components

Cloudwatch monitoring flow

Visualizing streaming data

Working with ElasticSearch using Kibana

Cloudwatch vs Elasticsearch

Elasticsearch visualization flow

An alternative approach

Another alternative approach

Monitoring and visualizing streaming data

Cosa alimenta, senza che ce ne accorgiamo, i sistemi che usiamo ogni giorno, come il rilevamento delle frodi che protegge le nostre transazioni o i semafori che tengono fluido il traffico? La risposta è lo streaming dei dati (dati generati in modo continuo da fonti diverse) e le tecnologie serverless—come Amazon Kinesis e AWS Lambda.
<br /> <br />
In questo corso imparerai a sfruttare queste tecnologie potenti aiutando una data engineer fittizia di nome Cody. Il tuo obiettivo è aiutarla a raccogliere dati in streaming in tempo reale dai veicoli di proprietà della città, analizzarli e inviare avvisi pertinenti, come segnalazioni di velocità, ai conducenti. Con Amazon Kinesis e Firehose imparerai a ingerire dati da milioni di fonti, per poi usare Kinesis Analytics per analizzare i dati mentre scorrono nello stream. Attiverai anche funzioni serverless in AWS Lambda che eseguiranno azioni in modo condizionale in base ai dati ricevuti. Al termine del corso saprai creare dashboard ElasticSearch in tempo reale con AWS QuickSight e CloudWatch—e, si spera, avrai aiutato Cody a portare a termine il suo ambizioso progetto.


Introduction to AWS Boto in Python

Introduction to Shell

Streaming Concepts

Scopri come gestire dati in streaming con tecnologie serverless su AWS.

Dati in streaming con AWS Kinesis e Lambda

Scopri come lavorare con i dati in streaming usando le tecnologie serverless su AWS.

Firehose scrive i record in arrivo nel bucket S3.

L'evento PutObject generato da S3 attiva la nostra funzione Lambda

Il Lambda Layer abilita l'uso di Pandas nel gestore Lambda

Il gestore della funzione Lambda viene eseguito e attiva una notifica SNS

Flusso Lambda

Dati in streaming con AWS Kinesis e Lambda

Esercizio pratico interattivo