LoginSignup
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

AI・機械学習 Qiita Tech Festa 2025
Qiita Tech Festa20252025年7月15日まで開催中！
@mkydk(Daiki Mukaiyama)

Amazon Kinesis Video Streamsの組み込み機能とKinesisファミリー連携による統合動画処理システムの構築

Posted at

概要

Amazon Kinesis Video Streamsの組み込み機能（HLS/DASH、WebRTC、GetClip API等）を活用し、他のKinesisサービスと連携することで、IoTデバイスからの動画データをリアルタイムに収集・処理・分析する統合システムの構築方法を解説します。実際のアーキテクチャパターンとコスト最適化手法も含めて紹介します。

目次

  1. はじめに
  2. Kinesisファミリーとの関連性
  3. Kinesis Video Streamsの組み込み機能詳細
  4. 実装例：IoTカメラシステムでの統合活用
  5. アーキテクチャパターンとベストプラクティス
  6. 料金体系と運用考慮事項
  7. 終わりに
  8. 参考文献・参考サイト

はじめに

動画データの処理需要が急速に拡大する中、AWSのKinesisファミリーは包括的なリアルタイムデータ処理ソリューションとして注目されています。特にAmazon Kinesis Video Streamsは、数百万のソース（スマートフォン、セキュリティカメラ、ウェブカメラ、車載カメラ、ドローンなど）から大量のライブ動画データを取り込むことができる強力なサービスです。

本記事では、Kinesis Video Streamsの組み込み機能を中心に、他のKinesisサービスとの連携による統合的な動画処理システムの構築方法を詳しく解説します。単体のサービス利用ではなく、Kinesisファミリー全体の連携による真の価値創出を目指します。

Kinesisファミリーとの関連性

Kinesisファミリー概要

AWSのKinesisファミリーは、用途に応じて4つの主要サービスに分かれています：

サービス名 主な用途 データ形式 リアルタイム性
Kinesis Video Streams 動画・音声データの収集・処理 非構造化（動画/音声） 低遅延～準リアルタイム
Kinesis Data Streams 構造化データのストリーミング 構造化データ リアルタイム
Kinesis Data Firehose データの変換・配信 構造化・半構造化 準リアルタイム
Kinesis Analytics ストリームデータの分析 構造化データ リアルタイム

image.png

連携パターンの特徴

Kinesis Video Streamsと他のKinesisサービスの連携では、以下の統合パターンが一般的です：

  1. メタデータ抽出パターン: Video Streamsから動画解析結果をData Streamsに送信
  2. リアルタイム分析パターン: 抽出されたデータをKinesis Analyticsで即座に分析
  3. 長期保存パターン: Firehoseを使用してS3やRedshiftへのデータ配信

これらの連携により、動画データから得られる洞察を構造化データとして他のシステムで活用できるようになります。

Kinesis Video Streamsの組み込み機能詳細

HLS（HTTP Live Streaming）出力機能

HLS機能は、GetHLSStreamingSessionURL APIを使用してストリーミングセッションを作成することで利用できます。この機能により、標準的なHTTPプロトコルでの動画配信が可能になります。

特徴：

  • Apple社が開発したストリーミングプロトコル
  • モバイルデバイスでの再生に最適化
  • 適応的ビットレート配信に対応

実装例：

// HLSストリーミングセッションURL取得
const params = {
    StreamName: 'my-video-stream',
    PlaybackMode: 'LIVE',
    HLSFragmentSelector: {
        FragmentSelectorType: 'SERVER_TIMESTAMP'
    }
};

const hlsUrl = await kinesisVideoArchivedMedia
    .getHLSStreamingSessionURL(params)
    .promise();

DASH（Dynamic Adaptive Streaming）出力機能

MPEG-DASH機能は、GetDASHStreamingSessionURL APIでストリーミングセッションを作成します。国際標準規格に基づく柔軟なストリーミング配信を実現します。

DASHの利点：

  • オープンスタンダード準拠
  • より細かい適応制御が可能
  • 最大10の同時MPEG-DASHストリーミングセッションをサポート

GetClip API機能

GetClip APIは指定された時間範囲のアーカイブされたオンデマンドメディアを含むクリップ（MP4ファイル）をダウンロードできる機能です。

活用シーン：

  • 異常検知時の証拠動画保存
  • 定期的なサンプル動画取得
  • アーカイブデータの後処理
# GetClip API使用例
import boto3
from datetime import datetime, timedelta

client = boto3.client('kinesis-video-archived-media')

# 過去1時間の動画クリップを取得
end_time = datetime.utcnow()
start_time = end_time - timedelta(hours=1)

response = client.get_clip(
    StreamName='surveillance-camera-01',
    ClipFragmentSelector={
        'FragmentSelectorType': 'SERVER_TIMESTAMP',
        'TimestampRange': {
            'StartTimestamp': start_time,
            'EndTimestamp': end_time
        }
    }
)

WebRTC双方向通信機能

Kinesis Video StreamsのWebRTC機能により、カメラIoTデバイス、ウェブブラウザ、モバイルデバイス間でのライブピアツーピアメディアストリーミングやリアルタイムなオーディオ・ビデオ対話が可能になります。

WebRTCの特徴：

  • 超低遅延通信（通常100ms以下）
  • P2P接続による効率的な帯域利用
  • セキュアな暗号化通信

Amazon Rekognition Video連携

Kinesis Video StreamsはAmazon Rekognition Videoと自動連携し、リアルタイムでの物体検出、顔認識、行動分析を実現できます。この連携により、動画から抽出された構造化データを他のKinesisサービスに流すことが可能になります。

実装例：IoTカメラシステムでの統合活用

システムアーキテクチャ

IoTセキュリティカメラシステムを例に、Kinesisファミリー全体を活用した統合システムを構築してみましょう。

構成要素：

  1. IoTカメラデバイス（Video Streams Producer）
  2. Kinesis Video Streams（動画データ収集）
  3. Lambda関数（Rekognition Video連携・メタデータ抽出）
  4. Kinesis Data Streams（構造化データストリーミング）
  5. Kinesis Analytics（リアルタイム異常検知）
  6. SNS（アラート通知）

image.png

セットアップ手順

1. Kinesis Video Streamの作成

# AWS CLIでVideo Stream作成
aws kinesisvideo create-stream \
    --stream-name "security-camera-stream" \
    --data-retention-in-hours 168 \
    --media-type "video/h264,audio/aac"

2. Data Stream作成（メタデータ用）

# メタデータ処理用のData Stream作成
aws kinesis create-stream \
    --stream-name "camera-metadata-stream" \
    --shard-count 2

3. Lambda関数による連携処理

import boto3
import json
from datetime import datetime

def lambda_handler(event, context):
    """
    Kinesis Video Streamsからの動画をRekognitionで分析し、
    結果をKinesis Data Streamsに送信
    """
    rekognition = boto3.client('rekognition')
    kinesis = boto3.client('kinesis')
    
    # Rekognition Video分析開始
    response = rekognition.start_label_detection(
        Video={
            'KinesisVideoStream': {
                'Arn': event['video_stream_arn'],
                'ConsumerArn': event['consumer_arn']
            }
        },
        NotificationChannel={
            'SNSTopicArn': 'arn:aws:sns:region:account:rekognition-topic',
            'RoleArn': 'arn:aws:iam::account:role/rekognition-role'
        }
    )
    
    # メタデータをData Streamsに送信
    metadata = {
        'timestamp': datetime.utcnow().isoformat(),
        'camera_id': event['camera_id'],
        'job_id': response['JobId'],
        'status': 'processing'
    }
    
    kinesis.put_record(
        StreamName='camera-metadata-stream',
        Data=json.dumps(metadata),
        PartitionKey=event['camera_id']
    )
    
    return {
        'statusCode': 200,
        'job_id': response['JobId']
    }

Kinesis Analyticsによるリアルタイム分析

-- 異常検知のためのSQL（Kinesis Analytics）
CREATE OR REPLACE STREAM "ANOMALY_DETECTION_STREAM" (
    camera_id VARCHAR(32),
    detected_objects INTEGER,
    confidence_avg DECIMAL(5,2),
    anomaly_score DECIMAL(5,2),
    event_time TIMESTAMP
);

CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "ANOMALY_DETECTION_STREAM"
SELECT STREAM 
    camera_id,
    COUNT(object_type) as detected_objects,
    AVG(confidence) as confidence_avg,
    CASE 
        WHEN COUNT(object_type) > 10 THEN 1.0
        WHEN AVG(confidence) < 0.5 THEN 0.8
        ELSE 0.0
    END as anomaly_score,
    ROWTIME as event_time
FROM "SOURCE_SQL_STREAM_001"
WHERE confidence > 0.3
GROUP BY camera_id, 
         RANGE (ROWTIME INTERVAL '1' MINUTE);

アーキテクチャパターンとベストプラクティス

代表的な統合アーキテクチャパターン

1. リアルタイム監視パターン

  • 用途: セキュリティ監視、品質管理
  • 特徴: 低遅延での異常検知とアラート
  • 推奨構成: Video Streams → Lambda → Data Streams → Analytics → SNS

2. データレイク統合パターン

  • 用途: 長期分析、機械学習モデル訓練
  • 特徴: 大容量データの効率的な保存と処理
  • 推奨構成: Video Streams → Firehose → S3 → Athena/EMR

3. ハイブリッド処理パターン

  • 用途: リアルタイム処理と長期分析の両立
  • 特徴: 柔軟性とコスト効率のバランス
  • 推奨構成: 上記2パターンの組み合わせ

パフォーマンス最適化のベストプラクティス

  1. シャード設計の最適化

    • Data Streamsのシャード数は処理量に応じて調整
    • パーティションキーの分散を考慮した設計

  2. バッファリング戦略

    • Lambda関数でのバッチ処理によるコスト削減
    • Firehoseのバッファサイズ調整

  3. リージョン選択

    • レイテンシ最小化のための適切なリージョン選択
    • データ主権とコンプライアンス要件の考慮

image.png

料金体系と運用考慮事項

Kinesisファミリーの料金比較

AWS公式サイト ( https://aws.amazon.com/kinesis/pricing/ ) によると、各サービスの料金体系は以下の通りです：

サービス 課金要素 東京リージョン料金目安
Video Streams データ取り込み量、保存期間 $0.0085/GB（取り込み）
Data Streams シャード時間、PUT/GET数 $0.015/シャード時間
Analytics KPU（処理ユニット）時間 $0.11/KPU時間
Firehose データ処理量 $0.029/GB

コスト最適化戦略

  1. データ保持期間の最適化

    • Video Streamsの保持期間を業務要件に合わせて設定
    • 不要なデータの早期削除によるストレージコスト削減

  2. 処理の効率化

    • バッチ処理によるLambda実行回数削減
    • 適切なシャード数でのコスト最適化

  3. Reserved Capacityの活用

    • 予測可能なワークロードでの予約利用
    • 長期的なコスト削減効果

実際の運用では、AWS Cost ExplorerやCloudWatchを活用して、継続的なコスト監視と最適化を行うことが重要です。

終わりに

Amazon Kinesis Video Streamsの組み込み機能と他のKinesisサービスとの連携により、従来では困難だった大規模な動画データのリアルタイム処理が現実的になりました。特に以下の点で大きなメリットを実現できます：

  • 統合性: 単一のAWSエコシステム内での完結した処理
  • スケーラビリティ: 需要に応じた自動スケーリング
  • コスト効率: 従量課金による無駄のない料金体系
  • 信頼性: AWSの堅牢なインフラストラクチャ

今後、IoTデバイスの普及とエッジコンピューティングの発展により、Kinesis Video Streamsの活用領域はさらに拡大することが予想されます。本記事で紹介したアーキテクチャパターンを参考に、皆様の要件に適した統合システムの構築を検討してみてください。

次のステップとして、実際の環境での検証や、Amazon SageMakerとの連携による機械学習パイプラインの構築なども視野に入れて取り組まれることをお勧めします。

参考文献・参考サイト

1
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?