1. Pinpoint 아키텍처 설계
1.1 기본 아키텍처
AWSTemplateFormatVersion: '2010-09-09'
Resources:
PinpointApplication:
Type: AWS::Pinpoint::App
Properties:
Name: MarketingAutomation
SMSChannel:
Type: AWS::Pinpoint::SMSChannel
Properties:
ApplicationId: !Ref PinpointApplication
Enabled: true
SenderId: YourCompany
EmailChannel:
Type: AWS::Pinpoint::EmailChannel
Properties:
ApplicationId: !Ref PinpointApplication
Enabled: true
FromAddress: marketing@yourcompany.com
Identity: !Ref EmailIdentity
EmailIdentity:
Type: AWS::Pinpoint::EmailIdentity
Properties:
EmailIdentity: marketing@yourcompany.com1.2 이벤트 추적 설정
import boto3
import json
def setup_event_stream():
pinpoint = boto3.client('pinpoint')
firehose = boto3.client('firehose')
# Kinesis Firehose 생성
delivery_stream = firehose.create_delivery_stream(
DeliveryStreamName='pinpoint-events',
DeliveryStreamType='DirectPut',
S3DestinationConfiguration={
'BucketARN': 'arn:aws:s3:::your-bucket',
'RoleARN': 'arn:aws:iam::your-account:role/firehose-role',
'BufferingHints': {
'SizeInMBs': 5,
'IntervalInSeconds': 300
},
'Prefix': 'pinpoint/events/'
}
)
# Pinpoint 이벤트 스트림 설정
pinpoint.put_event_stream(
ApplicationId='your-app-id',
WriteEventStream={
'DestinationStreamArn': delivery_stream['DeliveryStreamARN'],
'RoleArn': 'arn:aws:iam::your-account:role/pinpoint-events-role'
}
)2. 세그먼트 자동화 구현
2.1 동적 세그먼트 생성
import boto3
from datetime import datetime, timedelta
def create_dynamic_segment():
pinpoint = boto3.client('pinpoint')
response = pinpoint.create_segment(
ApplicationId='your-app-id',
WriteSegmentRequest={
'Name': f'HighValueCustomers_{datetime.now().strftime("%Y%m%d")}',
'SegmentGroups': {
'Groups': [
{
'Type': 'ANY',
'Dimensions': [
{
'Demographic': {
'Channel': {
'Values': ['SMS', 'EMAIL'],
'DimensionType': 'INCLUSIVE'
}
}
},
{
'Behavior': {
'Recency': {
'Duration': 'DAY_30',
'RecencyType': 'ACTIVE'
}
}
},
{
'UserAttributes': {
'CustomAttribute': {
'Values': ['Premium'],
'AttributeType': 'CONTAINS'
}
}
}
]
}
]
}
}
)
return response['SegmentResponse']['Id']3. 다채널 캠페인 구축
3.1 템플릿 기반 캠페인 생성
def create_multichannel_campaign():
pinpoint = boto3.client('pinpoint')
# SMS 템플릿 생성
sms_template = pinpoint.create_sms_template(
SMSTemplateRequest={
'Body': '안녕하세요 {{name}}님! {{product}} 할인 쿠폰이 발급되었습니다. 지금 확인해보세요!',
'DefaultSubstitutions': json.dumps({
'name': '고객',
'product': '프리미엄'
})
},
TemplateName='DiscountPromoSMS'
)
# 이메일 템플릿 생성
email_template = pinpoint.create_email_template(
EmailTemplateRequest={
'Subject': '{{name}}님을 위한 특별한 제안',
'HtmlPart': '''
<html>
<body>
<h1>안녕하세요 {{name}}님!</h1>
<p>{{product}} 상품에 대한 특별 할인 혜택을 받아보세요.</p>
<a href="{{link}}">자세히 보기</a>
</body>
</html>
''',
'DefaultSubstitutions': json.dumps({
'name': '고객',
'product': '프리미엄',
'link': 'https://yourcompany.com/promotion'
})
},
TemplateName='DiscountPromoEmail'
)
# 캠페인 생성
campaign = pinpoint.create_campaign(
ApplicationId='your-app-id',
WriteCampaignRequest={
'Name': 'MultiChannelPromo',
'SegmentId': 'your-segment-id',
'Schedule': {
'Frequency': 'ONCE',
'StartTime': (datetime.now() + timedelta(hours=1)).isoformat()
},
'MessageConfiguration': {
'SMSMessage': {
'MessageType': 'TRANSACTIONAL',
'TemplateId': sms_template['CreateTemplateMessageBody']['MessageBody']['Id']
},
'EmailMessage': {
'TemplateId': email_template['CreateTemplateMessageBody']['MessageBody']['Id']
}
}
}
)4. 실시간 분석 파이프라인
4.1 Lambda를 통한 실시간 처리
import json
import boto3
from datetime import datetime
def lambda_handler(event, context):
# Kinesis Firehose에서 전달된 이벤트 처리
records = []
for record in event['Records']:
payload = json.loads(record['kinesis']['data'])
# 이벤트 타입별 처리
if payload['event_type'] == '_campaign.send':
process_campaign_send(payload)
elif payload['event_type'] == '_campaign.response':
process_campaign_response(payload)
return {
'statusCode': 200,
'body': json.dumps('Successfully processed events')
}
def process_campaign_send(payload):
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('CampaignMetrics')
# 캠페인 전송 지표 업데이트
table.update_item(
Key={
'campaign_id': payload['campaign_id'],
'date': datetime.now().strftime('%Y-%m-%d')
},
UpdateExpression='ADD sends :inc',
ExpressionAttributeValues={
':inc': 1
}
)
def process_campaign_response(payload):
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('CampaignMetrics')
# 응답 유형별 처리
response_type = payload['attributes'].get('response_type')
update_expr = 'ADD '
if response_type == 'click':
update_expr += 'clicks :inc'
elif response_type == 'open':
update_expr += 'opens :inc'
table.update_item(
Key={
'campaign_id': payload['campaign_id'],
'date': datetime.now().strftime('%Y-%m-%d')
},
UpdateExpression=update_expr,
ExpressionAttributeValues={
':inc': 1
}
)5. 실전 구현 예제
5.1 A/B 테스트 캠페인
def create_ab_test_campaign():
pinpoint = boto3.client('pinpoint')
# A/B 테스트용 세그먼트 분할
segment_a = create_segment_subset('A그룹', 0.5)
segment_b = create_segment_subset('B그룹', 0.5)
# 각 변형별 캠페인 생성
campaign_a = create_variant_campaign('A', segment_a,
'지금 구매하시면 20% 할인!',
'LIMITED_TIME_OFFER')
campaign_b = create_variant_campaign('B', segment_b,
'오늘만 특가! 수량 한정 판매',
'SCARCITY')
return campaign_a, campaign_b
def create_segment_subset(name, percentage):
return pinpoint.create_segment(
ApplicationId='your-app-id',
WriteSegmentRequest={
'Name': name,
'SegmentGroups': {
'Groups': [
{
'Type': 'RANDOM',
'Dimensions': [
{
'Values': [str(percentage)],
'DimensionType': 'INCLUSIVE'
}
]
}
]
}
}
)6. 성능 최적화와 비용 관리
6.1 비용 모니터링 설정
def setup_cost_monitoring():
cloudwatch = boto3.client('cloudwatch')
# SMS 비용 알람 설정
cloudwatch.put_metric_alarm(
AlarmName='PinpointSMSCostAlarm',
ComparisonOperator='GreaterThanThreshold',
EvaluationPeriods=1,
MetricName='SMSMonthlySpend',
Namespace='AWS/Pinpoint',
Period=86400,
Statistic='Sum',
Threshold=1000.0,
AlarmDescription='SMS 비용이 일일 한도를 초과했습니다',
AlarmActions=[
'arn:aws:sns:region:account-id:alerting-topic'
]
)6.2 처리량 최적화
def optimize_throughput():
pinpoint = boto3.client('pinpoint')
# SMS 채널 설정 최적화
pinpoint.update_sms_channel(
ApplicationId='your-app-id',
SMSChannelRequest={
'Enabled': True,
'SenderId': 'YourCompany',
'ShortCode': 'your-short-code',
'DefaultMessageType': 'TRANSACTIONAL',
'RegisteredKeywords': ['STOP', 'UNSUBSCRIBE']
}
)
# 이메일 채널 설정 최적화
pinpoint.update_email_channel(
ApplicationId='your-app-id',
EmailChannelRequest={
'Enabled': True,
'FromAddress': 'marketing@yourcompany.com',
'Identity': 'your-ses-identity',
'ConfigurationSet': 'your-config-set'
}
)결론
AWS Pinpoint를 활용한 마케팅 자동화 시스템은 세그먼트 관리부터 다채널 캠페인 실행, 실시간 분석까지 포괄적인 기능을 제공합니다.
이를 통해 개인화된 고객 경험을 제공하고 마케팅 효과를 최적화할 수 있습니다.
에러가 나도 괜찮아 — 그건 내가 배우고 있다는 증거야.