In dit eerste hoofdstuk leer je het verschil tussen batch- en streamverwerking kennen, maak je je eerste stream, beheer je de rechten, en schrijf en lees je naar en van de stream.

Wat is streaming en waarom is het belangrijk?

Batch vs. stream

Producers en destinations

Firehose-delivery streams beheren

Klaarmaken voor de eerste stream

Rollen maken

IAM-gebruikers versus rollen

Oefenen met machtigingen

S3-bucket maken

Werken met de Firehose-deliverystream

Maak je eerste Firehose-stream

Schrijven naar een Firehose-stream

Firehose-gegevens lezen

Streamen in de cloud

De volgende stap in je streaming data-reis is leren hoe je transformerende lambda-functies gebruikt om serverless te gaan. In praktische oefeningen voeg je lambda-lagen toe en trigger je lambda-functies bij specifieke voorwaarden.

Serverless gaan

Lambda-functies maken en uitvoeren

Opties

Geschikt voor een server

Geschikt voor serverless

Serverless vs servers

Trigger

Handler

Layer

Onderdelen van een Lambda-functie

# 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
    }

Gegevens uit S3 lezen op basis van een event

Je eerste live lambda!

Een lambda-laag toevoegen

Lambda-flow

Omgevingsvariabelen gebruiken

Serverloze dataworkflow

Snel draaiende Lambdas

Lambdas die langer nodig hebben om te draaien

Snelle vs. langzame Lambdas

Tijdgetriggerde Lambda maken

Een tijdgetriggerde Lambda

Tijdgestuurde lambda-code

Serverless API's

Maak een door API getriggerde Lambda

Maak een API-lambda

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):
    is_async = event['____']['async']
    if is_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):
    is_async = event['queryStringParameters']['async']
    if is_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"
    }

Een serverloze API

Je bent nu klaar om streaming data te encoderen en decoderen en data direct in de stream te analyseren. Je gebruikt zelfs meerdere streams om dagelijkse topsnelheden van voertuigen te krijgen.

Een transformerende Lambda

Data transformeren binnen een stream

Coderen en decoderen van base64

Maak een transformationele Lambda

Een minimalistische transformationele lambda

Data analyseren in de stream

Een Kinesis Data Analytics-toepassing maken

Kinesis Data Analytics vs transformationele Lambda

Een Kinesis Data Analytics-toepassing bouwen

Kinesis Data Analytics SQL-componenten

Meerdere streams gebruiken

Data leveren vanuit Kinesis Analytics

Bepaal de dagelijkse topsnelheid

Kinesis Data Analytics gebruiken voor meldingen

Streaming data analyseren

In dit laatste hoofdstuk ontdek je hoe je de prestaties van je stream monitort met logs, metrische gegevens, alarmen en dashboards. Je gebruikt Elasticsearch en bouwt je eigen Kibana-dashboard.

Praktijkcase streaming data

Een Elasticsearch-cluster maken

RedShift vs Elasticsearch

Prestaties monitoren

CloudWatch-dashboards en -alarmen

CloudWatch-onderdelen

CloudWatch-monitoringflow

Streaminggegevens visualiseren

Werken met ElasticSearch in Kibana

CloudWatch vs Elasticsearch

Stroom voor Elasticsearch-visualisaties

Een alternatieve aanpak

Nog een alternatieve aanpak

Streaming data monitoren en visualiseren

Wat drijft systemen die we dagelijks gebruiken zonder dat we het merken, zoals fraudedetectie die onze transacties beveiligt of verkeerslichten die het verkeer soepel laten doorstromen? Het antwoord is streaming data (data die continu door verschillende bronnen wordt gegenereerd) en serverless-technologieën—zoals Amazon Kinesis en AWS Lambda.
<br /> <br />
In deze cursus leer je hoe je deze krachtige technologieën inzet door een fictieve data-engineer, Cody, te helpen. Je doel is om haar te helpen realtime streaming data van gemeentelijke voertuigen te verzamelen, de data te analyseren en relevante waarschuwingen—zoals snelheidsmeldingen—naar bestuurders te sturen. Met Amazon Kinesis en Firehose leer je data in te lezen van miljoenen bronnen, waarna je Kinesis Analytics gebruikt om data te analyseren terwijl die door de stream beweegt. Je zet ook serverless-functies op in AWS Lambda die voorwaardelijk acties starten op basis van de ontvangen data. Aan het einde van deze training weet je hoe je live ElasticSearch-dashboards maakt met AWS QuickSight en CloudWatch—en heb je hopelijk Cody geholpen haar ambitieuze project te voltooien.


Introduction to AWS Boto in Python

Introduction to Shell

Streaming Concepts

Leer werken met streaming data via serverloze technologieën op AWS.

Streaming data met AWS Kinesis en Lambda

Leer werken met streaming data via serverless technologieën op AWS.

Zijn geschikt voor asynchroon het triggeren van een andere functie

Hebben een lagere uitvoeringstijdlimiet nodig

Zijn geschikt om te triggeren via een API-request

Worden gebruikt om veel bestanden te lezen en te aggregeren

Kunnen afhangen van het synchrone resultaat van een andere lambda

Hebben mogelijk een hogere uitvoeringstijdlimiet nodig

Snelle vs. langzame Lambdas

Streaming data met AWS Kinesis en Lambda

Interactieve oefening met praktijkervaring