示例工程教程后端【免费下载链接】aws-doc-sdk-examplesWelcome to the AWS Code Examples Repository. This repo contains code examples used in the AWS documentation, AWS SDK Developer Guides, and more. For more information, see the Readme.md file below.项目地址https://gitcode.com/gh_mirrors/aw/aws-doc-sdk-examples点击查看免费下载导读本文围绕 AWS SDK 代码示例仓库中Amazon Data Firehose Common Actions 场景展开讲解如何通过PutRecord单条写入与PutRecordBatch批量写入两个核心 API把流式数据可靠地写入 Data Firehose 交付流Delivery Stream并配合 CloudFormation 基础设施、Faker 模拟数据与 CloudWatch 监控指标形成一套可在本地运行、可直接落地到生产项目的完整数据接入方案。读完本文你将掌握 Firehose 场景的部署命令、模拟数据生成方式、脚本规格含指数退避重试、批量分片、结构化日志与指标监控并能结合仓库中 Python SDK 的参考实现快速移植到任意语言。场景概览三个用户任务构成完整工作流本场景的目标是演示如何用各语言 AWS SDK 向 Data Firehose 交付流写入单条记录PutRecord与记录批次PutRecordBatch覆盖交付流的创建、配置与使用全链路。场景由三个用户任务组成详见 scenarios/features/firehose/README.md创建基础设施使用仓库提供的 CloudFormation 模板firehose-stack.yml创建交付流、S3 桶与 IAM 角色生成模拟数据运行 mock_data.py产出包含 5,550 条伪造网络记录的sample_records.json执行语言特定脚本依据 SPECIFICATION.md 实现并运行脚本观察记录写入与末尾的指标输出。说明仓库 README 与mock_data.py中关于数据量的描述存在差异——README 表述为 5,550 条而 mock_data.py 实际循环生成 2,550 条。实现脚本时以实际生成的文件为准脚本应能自适应处理任意数量的记录。关于 Amazon Data FirehoseAmazon Data Firehose 是向数据湖、数据存储与分析服务可靠加载流式数据的最简便方式。它能够捕获、转换并加载流式数据到 Amazon S3、Amazon Redshift、Amazon Elasticsearch Service 与 Splunk从而借助现有 BI 工具与仪表盘实现近实时分析。本场景中的交付流即采用DirectPut类型将数据直接落入 S3 目标桶。部署基础设施CloudFormation 模板深度解读模板三要素firehose-stack.yml 由三部分核心资源构成资源类型作用与关键配置MyFirehoseBucketAWS::S3::Bucket交付流的数据落地桶桶名通过!Join拼接my-firehose-bucket-与栈 ID 片段!Select [2, !Split [/, !Ref AWS::StackId]]保证全局唯一附带UpdateReplacePolicy: Delete与DeletionPolicy: Delete删除栈时同步清理桶MyFirehoseRoleAWS::IAM::Role服务角色信任策略中Principal指定firehose.amazonaws.comAction: sts:AssumeRole允许 Firehose 服务代入该角色MyFirehoseDeliveryStreamAWS::KinesisFirehose::DeliveryStream核心交付流DeliveryStreamType: DirectPut直接写入模式S3DestinationConfiguration将BucketARN与RoleARN指向上述桶与角色其中角色默认策略MyFirehoseRoleDefaultPolicy显式授予了写入 S3 桶所需的权限s3:AbortMultipartUpload、s3:GetBucketLocation、s3:GetObject、s3:ListBucket、s3:ListBucketMultipartUploads、s3:PutObject资源范围限定为桶本身及其下所有对象${MyFirehoseBucket.Arn}/*遵循最小权限原则。模板还导出了三个命名输出供后续脚本通过 CloudFormation 查询资源名FirehoseStreamName交付流名称BucketNameS3 桶名RoleArnIAM 角色 ARN部署命令在仓库根目录执行如下 CLI 命令创建栈aws cloudformation create-stack \ --stack-name FirehoseStack \ --template-body file://scenarios/features/firehose/resources/firehose-stack.yml \ --capabilities CAPABILITY_IAM--capabilities CAPABILITY_IAM必不可少——模板会创建 IAM 角色与策略CloudFormation 需要显式授权。部署成功后终端应输出所创建资源的名称与 ARN效果参见下图清理资源为避免持续计费实验结束后可删除栈CloudFormation 会一并清理桶、角色与交付流aws cloudformation delete-stack --stack-name FirehoseStack生成模拟数据mock_data.py 实现拆解mock_data.py 使用 Faker 库生成模拟网络告警记录。安装依赖Python ≥ 3.6pip install -r scenarios/features/firehose/resources/requirements.txt依赖仅一项faker见 requirements.txt。记录结构与生成逻辑每条记录包含三个字段ip_address由fake.ipv4()生成的随机 IPv4 地址timestampint(time.time())生成的当前 Unix 时间戳alert_level从[Low, Medium, High, Critical]随机取值的安全告警级别。脚本循环生成 2,550 条记录先打印前 5 条供预览再写入上级目录的sample_records.jsonjson.dump(records, f, indent2)并在结尾输出Generated 2,550 sample records.。运行命令python scenarios/features/firehose/resources/mock_data.py运行后仓库根目录将出现scenarios/features/firehose/sample_records.json。数据生成过程截图如下脚本规格SPECIFICATION五步实现指南SPECIFICATION.md 给出了实现语言无关脚本的完整规格共五步。1. 初始化 AWS SDK在所选语言中初始化 AWS SDK确保 IAM 角色/策略具备操作 Firehose 的权限确保第 2、3 节的基础设施与数据已生成可借助 CloudFormation 输出查询交付流名称。2. 定义配置参数定义delivery_stream_name、region、批大小与日志配置等参数。参考 Python 实现的config.py中get_config()返回的配置对象包含delivery_stream_name、region、sample_data_file等字段供客户端统一读取。3. 必需功能PutRecord 与 PutRecordBatch规格要求脚本按顺序执行两个函数先调用PutRecord逐条处理默认 100 条记录再用PutRecordBatch处理剩余记录合计完成全部数据处理。例如规格示例中为总计 5,550 条先 100 条单条、后 5,450 条批量。PutRecord API输入单条数据记录JSON 对象或纯文本及待处理记录数默认 10输出Firehose 服务返回的成功/失败响应错误处理实现带抖动jitter的指数退避重试。PutRecordBatch API输入记录列表输出逐条记录的成功/失败响应及批量记录数默认 500错误处理实现带抖动的指数退避重试可借助第三方库分页将大批量拆分为更小的块chunk批大小在吞吐与成本间权衡注意 Firehose 硬性限制——每请求最多 500 条记录或 4MB。4. 结构化日志捕获并记录 API 调用的成功与失败记录重试、异常、批量操作及调试/监控所需的关键信息实时输出脚本活动状态。5. 监控指标检查IncomingBytes与IncomingRecords指标确认有入站流量监控批操作的FailedPutCount。源码级参考Python SDK 实现与测试仓库提供了完整可跑的 Python 参考实现 firehose.py可作为任意语言移植的范本以下关键点与 SPECIFICATION 逐条对应。FirehoseClient客户端封装FirehoseClient在初始化时同时创建boto3.client(firehose)与boto3.client(cloudwatch)两个客户端region 取自配置为写入与指标监控复用同一连接上下文。PutRecord单条写入 指数退避put_record用backoff.on_exception(backoff.expo, Exception, max_tries5, jitterbackoff.full_jitter)装饰实现最多 5 次尝试的指数退避与全抖动重试对应 SPEC 的retries with exponential back-off and jitter。写入失败时记录Fail record日志并向上抛出异常。注意_create_record_entry会以 20% 概率模拟网络错误、以 10% 概率返回畸形 JSON——这是刻意设计用于让重试机制与错误日志在演示中真实触发。PutRecordBatch批量写入 分片put_record_batch同样带指数退避重试装饰内部以range(0, len(data), batch_size)将数据列表按默认 500 条切块每块构造{Data: json.dumps(record)}记录条目后调用firehose.put_record_batch对应 SPEC 的Pagination: Handle large batches by splitting them into smaller chunks与Batch Sizing。指标监控对接 CloudWatchget_metric_statistics以NamespaceAWS/Firehose、DimensionDeliveryStreamName查询 CloudWatchmonitor_metrics一次性拉取最近 10 分钟窗口的IncomingBytes、IncomingRecords、FailedPutCount三个指标IncomingBytes额外换算为 MB 输出无数据时提示No data found ... over the last 5 minutes。主流程编排__main__块按 SPEC 顺序执行load_sample_data读取sample_records.json→ 先对前 100 条循环调用put_record并monitor_metrics()→ 再对剩余记录调用put_record_batch(data[100:])并再次monitor_metrics()完整复现单条 → 批量 → 指标的演示节奏。测试验证test_firehose.py 使用moto的mock_firehose与mock_cloudwatch模拟 AWS 服务覆盖test_load_sample_data验证 JSON 文件读取test_put_record/test_put_record_batch验证单条与批量写入路径test_monitor_metrics验证 CloudWatch 指标查询test_no_credentials_error验证未配置凭证时抛NoCredentialsError。从源码结构看put_record_batch的异常分支except Exception as e会记录失败批次但不会中止整个流程实际使用中建议结合FailedPutCount返回值决定是否需要补偿重发。前置条件清单运行本场景前请确认AWS 账户拥有可用的 AWS 账户AWS CLI 已配置配置好具备 CloudFormation 创建权限的凭证Python ≥ 3.6安装并执行pip install -r scenarios/features/firehose/resources/requirements.txt可选Ailly配置 Ailly可用 start.sh 借助 AI 自动生成新的语言实现脚本。快速上手三步走# 1. 创建交付流等基础设施 aws cloudformation create-stack \ --stack-name FirehoseStack \ --template-body file://scenarios/features/firehose/resources/firehose-stack.yml \ --capabilities CAPABILITY_IAM # 2. 生成模拟数据 python scenarios/features/firehose/resources/mock_data.py # 3. 运行你的语言实现脚本观察记录写入与末尾指标输出 python python/example_code/firehose/scenarios/firehose-put-actions/firehose.py如何用 Ailly 自动生成新语言实现SPECIFICATION.md 与 start.sh 支持一种规格驱动 AI 生成的玩法配置 Aillynpm install -g ailly/cli并设置export AILLY_ENGINEbedrock后运行./scenarios/features/firehose/start.sh 语言名start.sh 会为指定语言创建目录依次生成02_SPECIFICATION.md、03_README.md、03_FILES.md、04_PROCESS.md与generate_files.py、run.sh其中03_README.md要求 LLM 输出面向新场景的 README含基础设施部署、数据生成、清理与多语言实现链接03_FILES.md让 LLM 以guru 级解决方案工程师身份列出构建/测试/运行所需的文件清单04_PROCESS.md要求 LLM 输出包裹在filename…/namecontents…/contents/file标签中的可执行代码generate_files.py用正则提取file块并落盘到app/目录run.sh串联执行。由此同一个规格可被自动扩展为 Java、Kotlin、Go 等多种 SDK 语言实现这也是本仓库以 SPEC 为中心、多语言交付的示例生成范式。场景运行效果与验证脚本执行过程中你会观察到单条写入阶段逐条打印Sent record: {...}或Fail record: {...}日志批量写入阶段按批次打印Successfully sent batch of 500 records或Failed to send N records in batch of 500末尾输出IncomingRecords、IncomingBytes含 MB 换算与FailedPutCount汇总指标。运行截图示意如下不同 SDK 输出略有差异关键设计要点回顾关注点规格要求参考实现位置单条写入PutRecord默认处理 100 条firehose.py批量写入PutRecordBatch默认批大小 500硬上限 500 条/4MBfirehose.py重试策略指数退避 抖动backoff.on_exception(... max_tries5, jitterbackoff.full_jitter)分页/分片大批量拆分为小块range(0, len(data), batch_size)结构化日志记录成功/失败/重试/异常_log_response/_log_batch_response指标监控IncomingBytes/IncomingRecords/FailedPutCountmonitor_metricsNamespaceAWS/Firehose基础设施S3 桶 IAM 角色 DirectPut 交付流firehose-stack.yml测试moto 模拟 Firehose/CloudWatchtest_firehose.py小结本场景用一条完整的基础设施 → 数据 → 脚本 → 指标链路示范了 Amazon Data Firehose 最常用的两个写入 API 的正确打开方式单条PutRecord适合低频精确写入批量PutRecordBatch面向高吞吐场景并须遵循 500 条/4MB 上限两者都建议配合指数退避重试与 CloudWatch 指标观测。无论你使用 Python、Java、Kotlin 还是其他语言都可以基于 SPECIFICATION.md 与 firehose.py 参考实现快速构建出生产级的数据接入脚本。赞分享示例工程教程后端【免费下载链接】aws-doc-sdk-examplesWelcome to the AWS Code Examples Repository. This repo contains code examples used in the AWS documentation, AWS SDK Developer Guides, and more. For more information, see the Readme.md file below.项目地址https://gitcode.com/gh_mirrors/aw/aws-doc-sdk-examples点击查看免费下载相关推荐AWS SDK for Java 2.x 实战使用 Amazon Data Firehose PutRecord 与 PutRecordBatch 批量写入 Delivery StreamAWS SDK for Java 2.x 实战使用 Amazon Data Firehose PutRecord 与 PutRecordBatch 批量写入示例工程教程后端使用 AWS SDK for Rust 向 Amazon Data Firehose 批量写入实时流数据PutRecordBatch 示例全解析使用 AWS SDK for Rust 向 Amazon Data Firehose 批量写入实时流数据PutRecordBatch 示例全解析 导读 本文以示例工程教程后端TensorRT 8 加速 EfficientDet 目标检测从 TensorFlow 权重到 TensorRT 引擎的完整实战指南TensorRT 8 加速 EfficientDet 目标检测从 TensorFlow 权重到 TensorRT 引擎的完整实战指南 本篇技术指南围绕仓库中的示例工程教程后端上一篇终极指南Locale-Emulator区域模拟器解决软件乱码和兼容性问题下一篇GitHub资源精准下载神器3分钟掌握DownGit的核心用法创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考