关键时间节点
AWS 经评估后决定终止 Amazon Pinpoint 服务，核心时间线如下：
- 2025-05-20 起：停止接受新客户。在此日期前注册的现有账户不受影响，可继续使用。
- 2026-10-30 起：正式终止支持。此后将无法访问 Pinpoint 控制台，也无法访问端点、细分、活动、旅程和分析等资源。
需要特别强调：短信、语音、移动推送、OTP 以及电话号码验证相关的 API 不受本次变更影响——这些接口已在 2024 年第三季度划归 AWS End User Messaging 继续提供服务。另外，本文为对官方公告的整理与精简，如与英文原文有出入，一律以原文为准。
你属于哪一类用户
Pinpoint 的客户大致分为两类，迁移目标也因此不同：
- 参与功能用户：使用端点（Endpoint）、细分（Segment）、营销活动（Campaign）、旅程（Journey）与分析等 engagement 能力。
- 消息渠道用户：使用短信、彩信/WhatsApp、推送、文字转语音等消息发送 API。
参与功能用户：迁移到 Amazon Connect
如果你使用的是参与功能，AWS 建议迁移到 Amazon Connect 客户主动参与方案（Amazon Connect Customer Profiles + Customer Outbound Campaigns）。它能在一个统一的应用里管理入站（如客服）与出站（如主动沟通），并支持跨渠道的个性化、及时互动与统一绩效跟踪。事件采集与移动分析则建议改用 Amazon Kinesis。
先选对迁移路径
动手之前，先判断你的工作负载特征：
| 如果你的工作负载… | 建议目标 |
|---|---|
| 包含语音 / SMS / 邮件 / WhatsApp，需要 AI 驱动的联络策略（预测式或渐进式外呼、客户细分）、坐席辅助的实时路由外呼，或在单一应用中协调入站与出站 | Amazon Connect Customer Outbound Campaigns |
| 超出 Connect 限额的吞吐需求，或需要保障送达 SLA，且无需坐席、无需 AI 决策、无需细分与编排 | AWS End User Messaging + Amazon SES |
两者结合的场景也很常见：坐席辅助互动用 Connect，大批量交易消息用 SES + End User Messaging。需要自动化迁移帮助的，可参考 AWS Marketplace 上的迁移工具。
迁移步骤
- 迁移端点与细分：Pinpoint 端点可建模为 Connect Customer Profiles，一个 Profile 最多容纳 3 个邮箱地址与 4 个电话号码。做法是创建一个不过滤的细分以覆盖全部端点，导出到 S3，再用 S3 连接器导入 Customer Profiles。若希望把多个端点聚合到同一客户档案下，可按 UserId 归并后再导入（见下文脚本）。
- 迁移渠道配置：在 Connect Customer 中按入门指引启用 SMS 与邮件通信。
- 迁移模板：Connect 使用与 Pinpoint 相同的 Handlebars 渲染引擎，但占位符写法不同。例如原来的
{{User.UserAttributes.PurchaseHistory}}需改为{{Attributes.Customer.Attributes.PurchaseHistory}}。先用get-email-template/get-sms-template等 API 取出模板，改完占位符后，用创建消息模板 API 在 Connect 中重建。 - 迁移营销活动：用
get-campaign取出定义，再用 Connect 的活动创建指南重建。 - 迁移旅程：用
get-journey取出定义，再用 Connect 客户旅程编排功能重建。
事件采集与移动分析
- Amplify SDK 用户：原本向 Pinpoint 发事件来更新端点、触发活动或分析应用使用情况的，迁移到 Kinesis，将事件流式传输到你的计算平台，由它去更新 Customer Profiles 并触发 Connect 活动。
- Put-Events 用户：原本只是把 Web / 移动端事件流式传到 Kinesis 的，可直接用 Amplify SDK 将事件送到 Kinesis。
暂时不支持的功能
- 应用内（In-App）消息
- 营销活动本身不支持推送（GCM / APNS / BAIDU 等）通知；但可在旅程中用 Lambda 动作 + Connect 推送模板发送
- 自定义渠道仅支持旅程，不支持营销活动
附：端点归并脚本
下面是一个将 Pinpoint 导出的端点 JSON 按 UserId 归并、转换为 Customer Profiles 格式并导入的参考脚本：
from collections import defaultdict
import json
def process_pinpoint_endpoints(input_file, output_file):
grouped_endpoints = defaultdict(list)
endpoints = []
with open(input_file, 'r') as file:
for line in file:
endpoints.append(json.loads(line))
for endpoint in endpoints:
user_id = endpoint.get('User', {}).get('UserId')
if user_id:
grouped_endpoints[user_id].append(endpoint)
customer_profiles = []
for user_id, user_endpoints in grouped_endpoints.items():
profile = {
'AccountNumber': user_id,
'Attributes': {},
'Address': {}
}
phone_numbers = set()
email_addresses = set()
output_dict = {}
for endpoint in user_endpoints:
attributes = endpoint.get('Attributes', {})
for key, value_list in attributes.items():
if len(value_list) == 1:
output_dict[key] = value_list[0]
else:
for i, item in enumerate(value_list):
output_dict[f"{key}_{i}"] = item
demographics = endpoint.get('Demographic') or {}
for key, value in demographics.items():
profile['Attributes'][f"Demographic_{key}"] = value
location = endpoint.get('Location', {})
profile['Address']['City'] = location.get('City')
profile['Address']['Country'] = location.get('Country')
profile['Address']['PostalCode'] = location.get('PostalCode')
profile['Address']['County'] = location.get('Region')
profile['Attributes']['Latitude'] = location.get('Latitude')
profile['Attributes']['Longitude'] = location.get('Longitude')
metrics = endpoint.get('Metrics', {})
for key, value in metrics.items():
profile['Attributes'][f"Metrics_{key}"] = str(value)
user = endpoint.get('User', {})
user_attributes = user.get('UserAttributes', {})
for key, value_list in user_attributes.items():
if len(value_list) == 1:
output_dict[key] = value_list[0]
else:
for i, item in enumerate(value_list):
output_dict[f"UserAttributes.{key}_{i}"] = item
profile['Attributes'].update(output_dict)
address = endpoint.get('Address')
if endpoint.get('ChannelType') in ('SMS', 'VOICE') and address:
phone_numbers.add(address)
if endpoint.get('ChannelType') == 'EMAIL' and address:
email_addresses.add(address)
for i, phone_number in enumerate(phone_numbers):
if i == 0:
profile['PhoneNumber'] = phone_number
elif i == 1:
profile['HomePhoneNumber'] = phone_number
elif i == 2:
profile['MobilePhoneNumber'] = phone_number
elif i == 3:
profile['BusinessPhoneNumber'] = phone_number
else:
profile['Attributes'][f"PhoneNumber_{i}"] = phone_number
for i, email_address in enumerate(email_addresses):
if i == 0:
profile['EmailAddress'] = email_address
elif i == 1:
profile['PersonalEmailAddress'] = email_address
elif i == 2:
profile['BusinessEmailAddress'] = email_address
else:
profile['Attributes'][f"EmailAddress_{i}"] = email_address
customer_profiles.append(profile)
with open(output_file, 'w') as f:
json.dump(customer_profiles, f, indent=2)
print(f"Processed {len(endpoints)} endpoints into {len(customer_profiles)} profiles.")
process_pinpoint_endpoints('pinpoint_endpoints.json', 'customer_profiles.json')
消息渠道用户：基本不受影响
短信、彩信/WhatsApp、推送、文字转语音等渠道已于 2024 年第三季度更名为 AWS End User Messaging，将继续提供服务，相关 API 的使用不受本次终止影响。
如果你用 Pinpoint 发送邮件，建议迁移到 Amazon SES；原本的邮件送达率控制面板，现在可在 SES 的 Virtual Deliverability Manager 中使用。
注销与数据导出
若想彻底删除 Pinpoint 数据，可直接调用 delete-app API 删除应用，再按指南清理模板。若只是导出归档，可参照以下方式：
- 端点：创建不过滤的细分导出到 S3 或本地。
- 细分 / 活动 / 旅程：分别用
get-segment/get-campaign/get-journey取出。 - 模板：用
list-templates列出后，再用get-email-template、get-sms-template、get-push-template、get-in-app-template逐个取出。 - 分析数据：开启事件数据流可导出后续原始事件；过去 3 个月的 KPI 可用
get-application-date-range-kpi、get-journey-date-range-kpi、get-campaign-date-range-kpi等系列 API 导出。需要删除 Mobile Analytics 应用的，可使用官方提供的 SigV4 签名 Python 脚本（需 Python 3.11+）。
小结
只要你的组织拥有一个 Pinpoint 账户，就可以继续使用其参与功能（细分、活动、旅程、分析、邮件）直到 2026-10-30 终止日。建议尽早评估工作负载特征，选择 Connect（参与场景）或 End User Messaging + SES（渠道场景）的迁移路径，并在截止日前完成数据与模板的导出归档，确保业务平滑过渡。
参考
官方迁移指南：Amazon Pinpoint 迁移文档
本文链接：https://bookshadow.com/weblog/2026/07/27/aws-pinpoint-end-of-support/
请尊重作者的劳动成果，转载请注明出处！书影博客保留对文章的所有权利。
捐赠书影博客
