In diesem ersten Kapitel lernst du die Unterschiede zwischen Batch- und Stream-Verarbeitung kennen, erstellst deinen ersten Stream, verwaltest seine Berechtigungen, schreibst in ihn und liest aus ihm.

Was ist Streaming und warum ist es wichtig?

Batch vs. Stream

Erzeuger und Bestimmungsorte

Verwaltung von Firehose-Lieferströmen

Vorbereitung auf den ersten Stream

Rollen erstellen

IAM Benutzer vs. Rollen

Praxis der Erlaubniserteilung

S3 Bucket erstellen

Arbeiten mit dem Firehose Lieferstrom

Erstelle deinen ersten Firehose Stream

Schreiben in einen Firehose-Stream

Firehose-Daten lesen

Streaming in der Cloud

Der nächste Schritt auf deiner Streaming-Data-Reise ist zu lernen, wie du transformative Lambda-Funktionen nutzen kannst, um serverlos zu werden. In praktischen Übungen fügst du Lambda-Ebenen hinzu und löst Lambda-Funktionen bei bestimmten Bedingungen aus.

Serverlos werden

Lambda-Funktionen erstellen und ausführen

Optionen

Großartig für einen Server

Großartig für Serverless

Serverlos vs. Server

Auslöser

Handler

Ebene

Komponenten der Lambda-Funktion

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

Lesen von Daten aus S3 auf der Grundlage eines Ereignisses

Dein erstes Live-Lambda!

Hinzufügen einer Lambda-Ebene

Lambda Flow

Lambda flow

Umgebungsvariablen verwenden

Serverloser Daten-Workflow

Schnell laufende Lambdas

Lambdas, die länger brauchen, um zu laufen

Schnelle vs. langsame Lambdas

Erstellen eines zeitgesteuerten Lambdas

Ein zeitgesteuertes Lambda

Zeitgesteuerter Lambda-Code

Serverlose APIs

API-gesteuerte Lambdas erstellen

Ein API-Lambda erstellen

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)

Eine serverlose API

Jetzt kannst du Streaming-Daten kodieren und dekodieren und die Daten direkt im Stream analysieren. Du wirst sogar mehrere Streams verwenden, um die täglichen Höchstgeschwindigkeiten der Fahrzeuge zu erhalten.

Ein transformatives Lambda

Daten innerhalb eines Streams umwandeln

Base64 kodieren und dekodieren

Erstellen eines Transformations-Lambdas

Ein einfaches Transformations-Lambda

Analysieren von Daten im Stream

Erstellen einer Kinesis-Datenanalyseanwendung

Kinesis Data Analytics vs. Transformational Lambda

Aufbau einer Kinesis-Datenanalyseanwendung

Kinesis data analytics SQL Komponenten

Mehrere Streams verwenden

Bereitstellung von Daten aus Kinesis Analytics

Erhalte die tägliche Höchstgeschwindigkeit

Verwendung von Kinesis-Datenanalysen für Warnungen

Analysieren von Streaming-Daten

In diesem letzten Kapitel erfährst du, wie du die Leistung deines Streams mithilfe von Protokollen, Metriken, Alarmen und Dashboards überwachen kannst. Du wirst Elasticsearch verwenden und dein eigenes Kibana-Dashboard erstellen.

Fallstudie zu Streaming-Daten

Erstellen eines Elasticsearch-Clusters

RedShift vs. Elasticsearch

Leistung überwachen

Cloudwatch dasbhoards und Alarme

Cloudwatch Komponenten

Cloudwatch Überwachungsablauf

Visualisierung von Streaming-Daten

Arbeiten mit ElasticSearch unter Verwendung von Kibana

Cloudwatch vs. Elasticsearch

Elasticsearch Visualisierungsablauf

Ein alternativer Ansatz

Ein anderer alternativer Ansatz

Überwachung und Visualisierung von Streaming-Daten

Was treibt die Systeme an, die wir jeden Tag benutzen, ohne es zu merken, wie z.B. die Betrugserkennung, die unsere Transaktionen sicher macht, oder die Verkehrssignale, die für einen reibungslosen Verkehrsfluss sorgen? Die Antwort sind Streaming-Daten (Daten, die kontinuierlich von verschiedenen Quellen erzeugt werden) und serverlose Technologien wie Amazon Kinesis und AWS Lambda.
<br /> <br />
In diesem Kurs lernst du, wie du diese leistungsstarken Technologien nutzen kannst, indem du einem fiktiven Dateningenieur namens Cody hilfst. Dein Ziel ist es, ihr dabei zu helfen, Echtzeit-Datenströme von stadteigenen Fahrzeugen zu sammeln, die Daten zu analysieren und relevante Warnungen wie Geschwindigkeitswarnungen an die Fahrer zu senden. Mit Amazon Kinesis und Firehose lernst du, wie du Daten aus Millionen von Quellen einlesen kannst, bevor du Kinesis Analytics nutzt, um die Daten zu analysieren, während sie sich durch den Stream bewegen. Du wirst auch serverlose Funktionen in AWS Lambda erstellen, die basierend auf den empfangenen Daten bedingte Aktionen auslösen. Am Ende dieser Schulung wirst du wissen, wie du mit AWS QuickSight und CloudWatch Live-Dashboards mit ElasticSearch erstellst - und hoffentlich Cody geholfen haben, ihr ehrgeiziges Projekt zu vollenden.


Introduction to AWS Boto in Python

Introduction to Shell

Streaming Concepts

Erfahren Sie, wie Sie mit serverlosen Technologien auf AWS Streaming-Daten verarbeiten.

Datenstreaming mit AWS Kinesis und Lambda

Lerne, wie du mit Streaming-Daten mithilfe von serverlosen Technologien auf AWS arbeitest.

Data Engineer

Adding a lambda layer

Let's practice!

Serverlos werden - Hinzufügen einer Lambda-Ebene