キューとストリーミング — リアルタイムデータ処理

# Kinesis ストリーム作成
aws kinesis create-stream \
  --stream-name buzz-events \
  --shard-count 10  # 10シャード = 10MB/s の書き込みキャパシティ
 
# 保持期間を7日に設定（デフォルトは24時間）
aws kinesis increase-stream-retention-period \
  --stream-name buzz-events \
  --retention-period-hours 168  # 7日間
 
# Enhanced Fan-Out を有効化（コンシューマーごとに2MB/sを保証）
aws kinesis register-stream-consumer \
  --stream-arn arn:aws:kinesis:ap-northeast-1:123456789:stream/buzz-events \
  --consumer-name buzz-analytics
 
# シャード数を増やす（スケールアップ）
aws kinesis update-shard-count \
  --stream-name buzz-events \
  --target-shard-count 20 \
  --scaling-type UNIFORM_SCALING

# Gemfile
gem 'aws-sdk-kinesis'

# app/services/event_publisher.rb
class EventPublisher
  KINESIS     = Aws::Kinesis::Client.new(region: 'ap-northeast-1')
  STREAM_NAME = 'buzz-events'
 
  # イベントを Kinesis に発行
  def self.publish(event_type:, data:, partition_key: nil)
    event = {
      event_type: event_type,
      event_id:   SecureRandom.uuid,
      timestamp:  Time.current.iso8601(3),  # ミリ秒精度
      version:    '1.0',
      data:       data
    }
 
    # partition_key でシャードを決定
    # 同じユーザーのイベントを同じシャードへ（順序保証）
    key = partition_key || data[:user_id]&.to_s || SecureRandom.hex
 
    KINESIS.put_record(
      stream_name:   STREAM_NAME,
      data:          JSON.generate(event),
      partition_key: key
    )
  rescue Aws::Kinesis::Errors::ProvisionedThroughputExceededException => e
    # スロットリング: 指数バックオフでリトライ
    Rails.logger.warn "Kinesis throttled, retrying: #{e.message}"
    Retryable.retryable(tries: 5, on: Aws::Kinesis::Errors::ProvisionedThroughputExceededException) do
      KINESIS.put_record(stream_name: STREAM_NAME, data: JSON.generate(event), partition_key: key)
    end
  end
 
  # バッチ発行（最大500件、コスト削減）
  def self.publish_batch(events)
    records = events.map do |event|
      {
        data:          JSON.generate(event),
        partition_key: event[:user_id]&.to_s || SecureRandom.hex
      }
    end
 
    records.each_slice(500) do |batch|
      response = KINESIS.put_records(
        stream_name: STREAM_NAME,
        records:     batch
      )
 
      # 失敗したレコードをリトライ
      if response.failed_record_count > 0
        failed = batch.zip(response.records).filter_map do |rec, result|
          rec if result.error_code
        end
        Rails.logger.warn "#{failed.count} records failed, retrying..."
        retry_batch(failed)
      end
    end
  end
 
  private
 
  def self.retry_batch(records, attempt: 0)
    return Rails.logger.error "Batch retry exhausted: #{records.count} records lost" if attempt >= 3
 
    sleep(2 ** attempt)  # 指数バックオフ: 1s, 2s, 4s
 
    response = KINESIS.put_records(stream_name: STREAM_NAME, records: records)
    if response.failed_record_count > 0
      retry_records = records.zip(response.records).filter_map { |r, res| r if res.error_code }
      retry_batch(retry_records, attempt: attempt + 1)
    end
  end
end

# モデルでイベント発行
class Post < ApplicationRecord
  after_commit :publish_kinesis_event, on: :create
 
  private
 
  def publish_kinesis_event
    EventPublisher.publish(
      event_type: 'post.created',
      data: {
        user_id:        user_id,
        post_id:        id,
        content_length: content.length,
        has_image:      image.attached?,
        hashtags:       extract_hashtags,
        language:       detect_language,
        created_at:     created_at.iso8601
      },
      partition_key: user_id.to_s
    )
  end
end
 
class Like < ApplicationRecord
  after_commit :publish_kinesis_event, on: [:create, :destroy]
 
  private
 
  def publish_kinesis_event
    EventPublisher.publish(
      event_type: destroyed? ? 'like.removed' : 'like.created',
      data: {
        user_id:  user_id,
        post_id:  post_id,
        actor_id: user_id,
        post_owner_id: post.user_id
      },
      partition_key: post_id.to_s  # 同じ投稿のいいねを同じシャードへ
    )
  end
end

# lambda/kinesis_processor/handler.py
import json
import base64
import boto3
from collections import defaultdict
 
dynamodb   = boto3.resource('dynamodb')
sqs        = boto3.client('sqs')
cloudwatch = boto3.client('cloudwatch')
 
NOTIFICATION_QUEUE = 'https://sqs.ap-northeast-1.amazonaws.com/123456789/buzz-notifications'
ACTIVITY_TABLE     = dynamodb.Table('buzz-activity-feed')
ANALYTICS_TABLE    = dynamodb.Table('buzz-analytics')
 
def lambda_handler(event, context):
    # バッチ処理（1つのLambda呼び出しで複数レコードを処理）
    processed = 0
    errors    = 0
 
    # イベントをタイプ別に集約（バッチ書き込みのため）
    likes_by_post  = defaultdict(int)
    activities     = []
    notifications  = []
 
    for record in event['Records']:
        try:
            payload    = json.loads(base64.b64decode(record['kinesis']['data']))
            event_type = payload['event_type']
            data       = payload['data']
 
            if event_type == 'post.created':
                handle_post_created(data, activities, notifications)
            elif event_type == 'like.created':
                likes_by_post[data['post_id']] += 1
                handle_like_created(data, activities, notifications)
            elif event_type == 'like.removed':
                likes_by_post[data['post_id']] -= 1
            elif event_type == 'follow.created':
                handle_follow_created(data, activities, notifications)
 
            processed += 1
        except Exception as e:
            print(f"Error processing record: {e}")
            errors += 1
 
    # バッチでDynamoDBに書き込み（コスト削減）
    if activities:
        batch_write_activities(activities)
 
    # バッチでSQSに通知メッセージを送信
    if notifications:
        batch_send_notifications(notifications)
 
    # いいね数をバッチ更新（DynamoDBのアトミック操作）
    for post_id, count in likes_by_post.items():
        if count != 0:
            update_like_count(post_id, count)
 
    # CloudWatchにメトリクスを送信
    cloudwatch.put_metric_data(
        Namespace='Buzz/Kinesis',
        MetricData=[
            {'MetricName': 'ProcessedRecords', 'Value': processed, 'Unit': 'Count'},
            {'MetricName': 'ErroredRecords',   'Value': errors,    'Unit': 'Count'}
        ]
    )
 
    return {'processed': processed, 'errors': errors}
 
 
def handle_like_created(data, activities, notifications):
    # アクティビティフィードに記録
    activities.append({
        'recipient_id': data['post_owner_id'],
        'activity_id':  generate_snowflake_id(),
        'actor_id':     data['user_id'],
        'action':       'like',
        'target_type':  'Post',
        'target_id':    data['post_id'],
        'created_at':   data.get('created_at'),
        'ttl':          int(time.time()) + 90 * 24 * 3600  # 90日後
    })
 
    # 通知キューへ
    notifications.append({
        'type':          'new_like',
        'recipient_id':  data['post_owner_id'],
        'actor_id':      data['user_id'],
        'post_id':       data['post_id']
    })
 
 
def batch_write_activities(activities):
    with ACTIVITY_TABLE.batch_writer() as batch:
        for activity in activities:
            batch.put_item(Item=activity)
 
 
def batch_send_notifications(notifications):
    # SQSに10件ずつバッチ送信（SQSの上限）
    for i in range(0, len(notifications), 10):
        batch = notifications[i:i+10]
        sqs.send_message_batch(
            QueueUrl=NOTIFICATION_QUEUE,
            Entries=[
                {
                    'Id':          str(idx),
                    'MessageBody': json.dumps(msg)
                }
                for idx, msg in enumerate(batch)
            ]
        )

# app/services/event_bridge_publisher.rb
class EventBridgePublisher
  EVENTS_CLIENT = Aws::EventBridge::Client.new(region: 'ap-northeast-1')
  EVENT_BUS_NAME = 'buzz-event-bus'
 
  EVENT_SOURCES = {
    posts:    'buzz.posts',
    social:   'buzz.social',
    users:    'buzz.users',
    payments: 'buzz.payments'
  }.freeze
 
  def self.publish(source:, detail_type:, detail:)
    EVENTS_CLIENT.put_events(
      entries: [{
        event_bus_name: EVENT_BUS_NAME,
        source:         EVENT_SOURCES.fetch(source),
        detail_type:    detail_type,
        detail:         JSON.generate(detail),
        time:           Time.current
      }]
    )
  end
 
  # バッチ発行（最大10件）
  def self.publish_batch(events)
    events.each_slice(10) do |batch|
      entries = batch.map do |event|
        {
          event_bus_name: EVENT_BUS_NAME,
          source:         EVENT_SOURCES.fetch(event[:source]),
          detail_type:    event[:detail_type],
          detail:         JSON.generate(event[:detail]),
          time:           Time.current
        }
      end
 
      response = EVENTS_CLIENT.put_events(entries: entries)
 
      if response.failed_entry_count > 0
        failed = response.entries.select { |e| e.error_code }
        Rails.logger.error "EventBridge: #{failed.count} events failed"
      end
    end
  end
end
 
# 使い方
EventBridgePublisher.publish(
  source:      :social,
  detail_type: 'UserFollowed',
  detail: {
    follower_id: current_user.id,
    followee_id: user.id,
    timestamp:   Time.current.iso8601
  }
)

# EventBridge ルール設定（CloudFormation）
NotificationRule:
  Type: AWS::Events::Rule
  Properties:
    EventBusName: buzz-event-bus
    EventPattern:
      source: ['buzz.social']
      detail-type:
        - 'UserFollowed'
        - 'PostLiked'
        - 'PostCommented'
        - 'UserMentioned'
    State: ENABLED
    Targets:
      - Arn: !GetAtt NotificationLambda.Arn
        Id: NotificationService
        # デッドレターキュー（失敗したイベントを保存）
        DeadLetterConfig:
          Arn: !GetAtt EventBridgeDLQ.Arn
        # リトライ設定
        RetryPolicy:
          MaximumRetryAttempts: 3
          MaximumEventAgeInSeconds: 3600
 
AnalyticsRule:
  Type: AWS::Events::Rule
  Properties:
    EventBusName: buzz-event-bus
    EventPattern:
      source: ['buzz.posts', 'buzz.social', 'buzz.users']
    State: ENABLED
    Targets:
      - Arn: !GetAtt AnalyticsKinesisStream.Arn
        Id: AnalyticsKinesis
        RoleArn: !GetAtt EventBridgeKinesisRole.Arn

# CloudFormation: Kinesis Firehose → S3 データレイク
BuzzFirehose:
  Type: AWS::KinesisFirehose::DeliveryStream
  Properties:
    DeliveryStreamName: buzz-events-firehose
    DeliveryStreamType: KinesisStreamAsSource
    KinesisStreamSourceConfiguration:
      KinesisStreamARN: !GetAtt BuzzKinesisStream.Arn
      RoleARN: !GetAtt FirehoseRole.Arn
 
    ExtendedS3DestinationConfiguration:
      BucketARN: !GetAtt DataLakeBucket.Arn
      RoleARN: !GetAtt FirehoseRole.Arn
 
      # 動的パーティショニング（年/月/日/イベントタイプ別）
      Prefix: "events/year=!{partitionKeyFromQuery:year}/month=!{partitionKeyFromQuery:month}/day=!{partitionKeyFromQuery:day}/event_type=!{partitionKeyFromQuery:event_type}/"
      ErrorOutputPrefix: "errors/year=!{timestamp:yyyy}/month=!{timestamp:MM}/!{firehose:error-output-type}/"
 
      DynamicPartitioningConfiguration:
        Enabled: true
        RetryOptions:
          DurationInSeconds: 300
 
      ProcessingConfiguration:
        Enabled: true
        Processors:
          - Type: MetadataExtraction
            Parameters:
              - ParameterName: MetadataExtractionQuery
                ParameterValue: '{year: .timestamp[0:4], month: .timestamp[5:7], day: .timestamp[8:10], event_type: .event_type}'
              - ParameterName: JsonParsingEngine
                ParameterValue: JQ-1.6
 
      BufferingHints:
        IntervalInSeconds: 300   # 5分ごとにS3に書き込み
        SizeInMBs: 128           # または128MB溜まったら書き込み
 
      CompressionFormat: GZIP   # gzip圧縮
 
      # Apache Parquet形式（列指向、Athenaで効率的）
      DataFormatConversionConfiguration:
        Enabled: true
        InputFormatConfiguration:
          Deserializer:
            OpenXJsonSerDe: {}
        OutputFormatConfiguration:
          Serializer:
            ParquetSerDe:
              Compression: SNAPPY

-- Amazon Athena でイベントデータを分析
-- S3のデータをSQLで直接クエリ（サーバーレス、低コスト）
 
-- テーブル作成（S3のデータを参照）
CREATE EXTERNAL TABLE buzz_events (
  event_type  STRING,
  event_id    STRING,
  timestamp   STRING,
  data        STRUCT<
    user_id:     BIGINT,
    post_id:     BIGINT,
    likes_count: INT
  >
)
PARTITIONED BY (
  year       STRING,
  month      STRING,
  day        STRING,
  event_type STRING
)
STORED AS PARQUET
LOCATION 's3://buzz-data-lake/events/'
TBLPROPERTIES ('parquet.compression' = 'SNAPPY');
 
-- パーティション情報を更新
MSCK REPAIR TABLE buzz_events;
 
-- 時間帯別のいいね数分析
SELECT
  DATE_TRUNC('hour', PARSE_DATETIME(timestamp, 'yyyy-MM-dd''T''HH:mm:ss.SSSZ')) as hour,
  COUNT(*) as total_likes,
  COUNT(DISTINCT data.user_id) as unique_users
FROM buzz_events
WHERE year = '2024'
  AND month = '12'
  AND event_type = 'like.created'
GROUP BY 1
ORDER BY 1;
 
-- コスト: $5/TBスキャン（Parquet + パーティショニングで大幅削減）

# app/channels/activity_channel.rb
class ActivityChannel < ApplicationCable::Channel
  def subscribed
    reject unless current_user
 
    stream_for current_user
    # Redis PubSub経由でブロードキャスト
 
    # 接続時に未読通知数を送る
    transmit({
      type:          'unread_count',
      unread_count:  current_user.unread_notifications_count
    })
  end
 
  def unsubscribed
    stop_all_streams
  end
 
  def mark_read(data)
    notification_ids = data['notification_ids']
    Notification.where(id: notification_ids, recipient_id: current_user.id)
                .update_all(read_at: Time.current)
  end
end
 
# Lambda から Rails に WebSocket メッセージを送る
# app/jobs/broadcast_activity_job.rb
class BroadcastActivityJob < ApplicationJob
  queue_as :critical
 
  def perform(recipient_id:, activity:)
    user = User.find(recipient_id)
    ActivityChannel.broadcast_to(user, {
      type:     'new_activity',
      activity: activity
    })
  end
end

// app/javascript/channels/activity_channel.js
import consumer from "channels/consumer"
 
const activityChannel = consumer.subscriptions.create("ActivityChannel", {
  connected() {
    console.log("ActivityChannel connected");
  },
 
  disconnected() {
    console.log("ActivityChannel disconnected");
    // 再接続を試みる
    setTimeout(() => this.consumer.connect(), 3000);
  },
 
  received(data) {
    switch (data.type) {
      case 'new_activity':
        this.handleNewActivity(data.activity);
        break;
      case 'unread_count':
        this.updateBadge(data.unread_count);
        break;
    }
  },
 
  handleNewActivity(activity) {
    // 通知バッジの数字を増やす
    const badge = document.getElementById('notification-badge');
    const count = parseInt(badge.dataset.count || '0') + 1;
    badge.dataset.count = count;
    badge.textContent  = count > 99 ? '99+' : count;
    badge.classList.remove('hidden');
 
    // トースト通知を表示
    this.showToast(this.formatActivityMessage(activity));
  },
 
  updateBadge(count) {
    const badge = document.getElementById('notification-badge');
    if (count > 0) {
      badge.textContent = count > 99 ? '99+' : count;
      badge.classList.remove('hidden');
    } else {
      badge.classList.add('hidden');
    }
  },
 
  showToast(message) {
    const toast = document.createElement('div');
    toast.className = 'toast-notification';
    toast.textContent = message;
    document.body.appendChild(toast);
    setTimeout(() => toast.remove(), 4000);
  },
 
  formatActivityMessage(activity) {
    const messages = {
      like:    `${activity.actor.username}さんがいいねしました`,
      follow:  `${activity.actor.username}さんがフォローしました`,
      comment: `${activity.actor.username}さんがコメントしました`
    };
    return messages[activity.action] || '新しいアクティビティ';
  }
});
 
export default activityChannel;