在本章中，您将了解批处理与流处理的区别，创建您的第一个数据流，管理其权限，并向数据流写入与读取数据。

什么是流式处理？为什么重要？

批处理 vs 流处理

生产者与目标

管理 Firehose Delivery Streams

准备创建第一个流

创建角色

IAM 用户与角色

权限练习

创建 S3 存储桶

使用 Firehose 传输流

创建您的第一个 Firehose 流

写入 Firehose 流

读取 Firehose 数据

云端流处理

在流数据学习之旅的下一步，您将学习使用具有转换能力的 Lambda 函数来实现无服务器架构。通过动手练习，您将添加 Lambda 层，并在满足特定条件时触发 Lambda 函数。

无服务器架构入门

创建并运行 Lambda 函数

选项

适合服务器

适合 Serverless

Serverless 与传统服务器

触发器

处理程序

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

基于事件从 S3 读取数据

您的第一个实时 Lambda！

添加 Lambda 层

Lambda 流程

使用环境变量

无服务器数据工作流

快速运行的 Lambda

运行时间更长的 Lambda

快速与慢速 Lambda

创建按时间触发的 Lambda

按时间触发的 Lambda

按时间触发的 Lambda 代码

无服务器 API

创建由 API 触发的 Lambda

创建一个 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)

走向无服务器

现在，您已准备好对流数据进行编码与解码，并直接在数据流中分析数据。您甚至会使用多个数据流来获取车辆的每日最高车速。

一个用于转换的 Lambda

在流中转换数据

编码与解码 base64

创建一个转换型 Lambda

一个精简的转换型 lambda

在流中分析数据

创建 Kinesis Data Analytics 应用程序

Kinesis Data Analytics 与转换型 Lambda 的对比

构建 Kinesis 数据分析应用

Kinesis Data Analytics 的 SQL 组件

使用多个流

从 Kinesis Analytics 交付数据

获取每日最高车速

使用 Kinesis 数据分析发送告警

分析流数据

在最后一章中，您将学习如何通过日志、指标、告警和仪表板来监控数据流的性能。您将使用 Elasticsearch，并构建自己的 Kibana 仪表板。

流式数据案例研究

创建 Elasticsearch 集群

RedShift 与 Elasticsearch

监控性能

CloudWatch 仪表板与告警

CloudWatch 组件

CloudWatch 监控流程

可视化流式数据

使用 Kibana 操作 ElasticSearch

CloudWatch 与 Elasticsearch

Elasticsearch 可视化流程

另一种实现思路

另一种可选方案

监控与可视化流数据

是什么在悄然驱动我们日常使用的系统，例如保障交易安全的欺诈检测，或让交通顺畅的信号控制？答案是流数据（由不同来源持续产生的数据）以及无服务器技术——如 Amazon Kinesis 和 AWS Lambda。
<br /> <br />
在本课程中，您将通过协助一位名叫 Cody 的虚构数据工程师来学习如何运用这些强大的技术。您的目标是帮助她收集来自城市公务车辆的实时流数据，对数据进行分析，并向驾驶员发送相关告警，例如超速提醒。借助 Amazon Kinesis 和 Firehose，您将学习如何从数百万个来源摄取数据；随后使用 Kinesis Analytics 在数据流动过程中进行分析。您还将使用 AWS Lambda 启动无服务器函数，根据接收到的数据有条件地触发相应操作。完成本课程后，您将掌握如何使用 AWS QuickSight 和 CloudWatch 创建实时的 ElasticSearch 仪表板——也希望能帮助 Cody 完成她的雄心勃勃的项目。


基于事件从 S3 读取数据

使用 AWS Kinesis 与 Lambda 处理流数据

动手互动练习