I det här första kapitlet lär du dig skillnaderna mellan batchbearbetning och strömbearbetning. Du skapar din första ström, hanterar dess behörigheter och skriver till och läser från den.

Vad är streaming och varför är det viktigt?

Batch vs. ström

Producenter och destinationer

Hantera Firehose delivery streams

Förbered dig för den första streamen

Skapa roller

IAM-användare vs. roller

Behörighetsövning

Skapa en S3-bucket

Arbeta med Firehose-leveransströmmen

Skapa ditt första Firehose-flöde

Skriva till en Firehose-ström

Läsa Firehose-data

Strömning i molnet

Nästa steg på din resa med strömdata är att lära dig använda transformerande lambdafunktioner för att arbeta serverlöst. Genom praktiska övningar lägger du till lambda-lager och utlöser lambdafunktioner under specifika villkor.

Dags att gå serverlöst

Skapa och kör Lambda-funktioner

Alternativ

Passar för server

Passar för serverlöst

Serverlöst vs servrar

Trigger

Handler

Layer

Komponenter i en lambdafunktion

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

Läsa data från S3 baserat på en händelse

Din första live-lambda!

Lägga till ett lambda-lager

Lambda-flödet

Använda miljövariabler

Serverlöst dataarbetsflöde

Snabba lambdafunktioner

Lambdafunktioner som tar längre tid att köra

Snabba och långsamma lambdafunktioner

Skapa en tidsutlöst Lambda-funktion

En tidsutlöst Lambda-funktion

Tidsutlöst lambda-kod

Serverlösa API:er

Skapa en API-triggad lambda

Skapa en 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):
    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)

Ett serverlöst API

Serverlös arkitektur

Nu är du redo att koda och avkoda strömdata samt analysera data direkt i strömmen. Du använder även flera strömmar för att ta fram fordonens dagliga topphastigheter.

En transformerande Lambda

Transformera data i en ström

Koda och avkoda base64

Skapa en transformerande Lambda-funktion

En enkel transformations-lambda

Analysera data i strömmen

Skapa en Kinesis-applikation för dataanalys

Kinesis Data Analytics vs transformations-Lambda

Bygg en Kinesis Data Analytics-applikation

SQL-komponenter i Kinesis Data Analytics

Använda flera strömmar

Leverera data från Kinesis Analytics

Hämta dagens topphastighet

Använda Kinesis Data Analytics för aviseringar

Analysera strömdata

I det här sista kapitlet får du lära dig hur du övervakar strömmens prestanda med hjälp av loggar, mätvärden, larm och dashboards. Du använder Elasticsearch och bygger din egen Kibana-dashboard.

Fallstudie: strömmande data

Skapa ett Elasticsearch-kluster

RedShift vs Elasticsearch

Övervaka prestanda

CloudWatch-instrumentpaneler och larm

CloudWatch-komponenter

Övervakningsflöde i CloudWatch

Visualisering av strömmande data

Arbeta med ElasticSearch via Kibana

CloudWatch vs Elasticsearch

Visualiseringsflöde i Elasticsearch

Ett alternativt tillvägagångssätt

Ett annat alternativt tillvägagångssätt

Övervaka och visualisera strömdata

Vad driver de system vi använder varje dag utan att tänka på det – som bedrägeriskydd för att hålla våra transaktioner säkra, eller trafiksignaler som får trafiken att flyta smidigt? Svaret är strömdata (data som kontinuerligt genereras från olika källor) och serverlös teknik – som Amazon Kinesis och AWS Lambda.
<br /> <br />
I den här kursen får du lära dig att använda dessa kraftfulla tekniker genom att hjälpa en fiktiv dataingenjör som heter Cody. Målet är att hjälpa henne att samla in realtidsströmdata från kommunala fordon, analysera datan och skicka relevanta varningar – till exempel hastighetsvarningar till förare. Med Amazon Kinesis och Firehose lär du dig att ta emot data från miljontals källor, och sedan använda Kinesis Analytics för att analysera data medan den rör sig genom strömmen. Du skapar även serverlösa funktioner i AWS Lambda som villkorsstyrt utlöser åtgärder baserat på inkommande data. I slutet av kursen vet du hur man skapar live-dashboards i ElasticSearch med AWS QuickSight och CloudWatch – och förhoppningsvis har du också hjälpt Cody att slutföra sitt ambitiösa projekt.


Introduction to AWS Boto in Python

Introduction to Shell

Streaming Concepts

Lär dig hantera strömmande data med serverlösa teknologier på AWS.

Strömdata med AWS Kinesis och Lambda

Lär dig arbeta med strömmande data med serverlösa tekniker på AWS.

Skapa funktionen i AWS Lambda-gränssnittet.

Lägg till utlösare: EventBridge (CloudWatch Events)

En tidsutlöst Lambda-funktion

Strömdata med AWS Kinesis och Lambda

Interaktiv övning med praktiskt arbete