Backend & APIs Topic
Video Streaming System Design
System Requirements & Scale
Functional Requirements
-
Video Upload: Large file uploads, transcoding, storage
-
Video Streaming: Adaptive bitrate streaming, low latency
-
Content Management: Metadata, thumbnails, categorization
-
User Experience: Watch history, playlists, likes, comments
-
Recommendations: Personalized content suggestions
-
Social Features: Subscriptions, sharing, notifications
-
Monetization: Ads, subscriptions, premium content
-
Analytics: View counts, engagement metrics, user behavior
-
Content Moderation: Copyright detection, inappropriate content
-
Live Streaming: Real-time broadcasting, chat
Non-Functional Requirements
-
Scalability: Handle 1B+ daily users, 1M+ concurrent streams
-
Availability: 99.99% uptime for critical services
-
Performance: < 1s video start time, < 200ms seek time
-
Quality: Adaptive bitrate, 4K/8K support, HDR
-
Global Reach: Low latency worldwide via CDN
-
Cost Efficiency: Optimize storage and bandwidth costs
-
Security: DRM, content protection, user data privacy
-
Reliability: No single point of failure, disaster recovery
High-Level Architecture
text
┌─────────────────────────────────────────────────────────────────────────┐
│ Client Applications │
├─────────────────────────────────────────────────────────────────────────┤
│ • Web (React/Next.js) • Mobile Apps (iOS/Android) │
│ • Smart TVs • Gaming Consoles │
│ • Desktop Apps • Set-top Boxes │
│ • Embedded Devices • Third-party Integrations │
└─────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────┐
│ API Gateway & Load Balancers │
├─────────────────────────────────────────────────────────────────────────┤
│ • Global Load Balancing (Anycast) • API Gateway (Kong/Envoy) │
│ • DDoS Protection • Rate Limiting │
│ • Authentication • Request Routing │
└─────────────────────────────────────────────────────────────────────────┘
│
┌───────────────────┼───────────────────┐
│ │ │
▼ ▼ ▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ Upload │ │ Streaming │ │ Metadata │
│ Pipeline │ │ Service │ │ Service │
└─────────────────┘ └─────────────────┘ └─────────────────┘
│ │ │
└───────────────────┼───────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────┐
│ Core Infrastructure │
├─────────────────────────────────────────────────────────────────────────┤
│ • CDN (CloudFront, Akamai) • Object Storage (S3, GCS) │
│ • Transcoding Cluster (FFmpeg) • Database (Cassandra, PostgreSQL) │
│ • Cache (Redis, Memcached) • Message Queue (Kafka, RabbitMQ) │
│ • Search (Elasticsearch) • Analytics (Spark, Flink) │
└─────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────┐
│ External Services │
├─────────────────────────────────────────────────────────────────────────┤
│ • Payment Gateways • Email/SMS • Content ID (Audible Magic) │
│ • Ad Servers • Analytics • DRM Providers (Widevine) │
│ • CDN Providers • Monitoring • Cloud Providers │
└─────────────────────────────────────────────────────────────────────────┘
Detailed Component Design
1. Video Upload Service
python
# services/upload_service.py
class VideoUploadService:
"""Handles video uploads with resume support"""
def __init__(self):
self.s3 = S3Client()
self.kafka = KafkaProducer()
self.redis = RedisCluster()
self.db = PostgreSQL()
async def initiate_upload(self, user_id: str, metadata: VideoMetadata) -> UploadSession:
"""Initiate multi-part upload session"""
# Generate unique video ID
video_id = self.generate_video_id()
# Create upload session in Redis
session_id = str(uuid.uuid4())
session = {
'session_id': session_id,
'video_id': video_id,
'user_id': user_id,
'file_size': metadata.file_size,
'chunk_size': 10 * 1024 * 1024, # 10MB chunks
'total_chunks': ceil(metadata.file_size / (10 * 1024 * 1024)),
'uploaded_chunks': set(),
'status': 'INITIATED',
'created_at': datetime.utcnow().isoformat(),
'expires_at': (datetime.utcnow() + timedelta(hours=24)).isoformat()
}
# Store session
await self.redis.setex(
f"upload_session:{session_id}",
86400, # 24 hours
json.dumps(session)
)
# Create S3 multi-part upload
s3_upload = await self.s3.create_multipart_upload(
bucket='video-uploads',
key=f"{video_id}/original"
)
# Update session with S3 upload ID
session['s3_upload_id'] = s3_upload['UploadId']
await self.redis.setex(
f"upload_session:{session_id}",
86400,
json.dumps(session)
)
# Create video record in database
await self.db.execute("""
INSERT INTO videos (
id, user_id, title, description,
privacy_status, upload_status, created_at
) VALUES (%s, %s, %s, %s, %s, %s, NOW())
""", (
video_id, user_id, metadata.title,
metadata.description, metadata.privacy,
'UPLOADING'
))
return UploadSession(
session_id=session_id,
video_id=video_id,
upload_urls=self._generate_presigned_urls(
video_id, session['total_chunks']
),
chunk_size=session['chunk_size']
)
async def upload_chunk(self, session_id: str, chunk_number: int,
chunk_data: bytes) -> bool:
"""Upload a single chunk"""
# Get session
session_data = await self.redis.get(f"upload_session:{session_id}")
if not session_data:
raise UploadSessionExpired()
session = json.loads(session_data)
# Upload chunk to S3
etag = await self.s3.upload_part(
bucket='video-uploads',
key=f"{session['video_id']}/original",
upload_id=session['s3_upload_id'],
part_number=chunk_number,
data=chunk_data
)
# Update session
session['uploaded_chunks'].add(chunk_number)
session['etags'] = session.get('etags', {})
session['etags'][chunk_number] = etag
# Check if upload is complete
if len(session['uploaded_chunks']) == session['total_chunks']:
session['status'] = 'COMPLETED'
# Complete S3 upload
parts = [
{'ETag': etag, 'PartNumber': num}
for num, etag in session['etags'].items()
]
await self.s3.complete_multipart_upload(
bucket='video-uploads',
key=f"{session['video_id']}/original",
upload_id=session['s3_upload_id'],
multipart_upload={'Parts': parts}
)
# Trigger processing pipeline
await self.kafka.produce('video_uploads', {
'event_id': str(uuid.uuid4()),
'event_type': 'UPLOAD_COMPLETED',
'video_id': session['video_id'],
'user_id': session['user_id'],
'timestamp': datetime.utcnow().isoformat()
})
# Update session in Redis
await self.redis.setex(
f"upload_session:{session_id}",
86400,
json.dumps(session)
)
return True
async def resume_upload(self, session_id: str) -> UploadSession:
"""Resume interrupted upload"""
# Get session
session_data = await self.redis.get(f"upload_session:{session_id}")
if not session_data:
raise UploadSessionExpired()
session = json.loads(session_data)
# Get uploaded chunks from S3
uploaded_parts = await self.s3.list_parts(
bucket='video-uploads',
key=f"{session['video_id']}/original",
upload_id=session['s3_upload_id']
)
# Update session with uploaded parts
uploaded_chunks = {part['PartNumber'] for part in uploaded_parts}
session['uploaded_chunks'] = uploaded_chunks
# Generate new upload URLs for missing chunks
missing_chunks = set(range(1, session['total_chunks'] + 1)) - uploaded_chunks
upload_urls = self._generate_presigned_urls(
session['video_id'], list(missing_chunks)
)
# Update session
await self.redis.setex(
f"upload_session:{session_id}",
86400,
json.dumps(session)
)
return UploadSession(
session_id=session_id,
video_id=session['video_id'],
upload_urls=upload_urls,
chunk_size=session['chunk_size'],
uploaded_chunks=list(uploaded_chunks)
)
def _generate_presigned_urls(self, video_id: str, chunks: List[int]) -> List[str]:
"""Generate presigned URLs for chunk uploads"""
urls = {}
for chunk in chunks:
url = self.s3.generate_presigned_url(
bucket='video-uploads',
key=f"{video_id}/chunks/{chunk}",
method='PUT',
expires_in=3600
)
urls[chunk] = url
return urls
def generate_video_id(self) -> str:
"""Generate unique video ID (Base64 encoded timestamp + random)"""
timestamp = int(time.time() * 1000)
random_bytes = os.urandom(6)
combined = struct.pack('>Q', timestamp)[2:] + random_bytes
return base64.urlsafe_b64encode(combined).decode().rstrip('=')
2. Video Processing Pipeline
python
# services/processing_service.py
class VideoProcessingService:
"""Video transcoding and processing pipeline"""
def __init__(self):
self.ffmpeg_cluster = FFMpegCluster()
self.s3 = S3Client()
self.kafka = KafkaProducer()
self.redis = RedisCluster()
self.db = PostgreSQL()
async def process_video(self, video_id: str):
"""Process uploaded video through pipeline"""
# Get video metadata
video = await self.db.execute("""
SELECT * FROM videos WHERE id = %s
""", (video_id,))
# Download original video
original_path = await self.s3.download_file(
bucket='video-uploads',
key=f"{video_id}/original",
local_path=f"/tmp/{video_id}/original"
)
# Extract video metadata
metadata = await self._extract_metadata(original_path)
# Update database with metadata
await self.db.execute("""
UPDATE videos
SET duration = %s, width = %s, height = %s,
framerate = %s, codec = %s, bitrate = %s,
updated_at = NOW()
WHERE id = %s
""", (
metadata['duration'], metadata['width'],
metadata['height'], metadata['framerate'],
metadata['codec'], metadata['bitrate'],
video_id
))
# Generate thumbnail
thumbnails = await self._generate_thumbnails(original_path, video_id)
# Create multiple quality versions
qualities = [
{'name': '144p', 'height': 144, 'bitrate': '200k'},
{'name': '240p', 'height': 240, 'bitrate': '400k'},
{'name': '360p', 'height': 360, 'bitrate': '800k'},
{'name': '480p', 'height': 480, 'bitrate': '1200k'},
{'name': '720p', 'height': 720, 'bitrate': '2500k'},
{'name': '1080p', 'height': 1080, 'bitrate': '5000k'},
{'name': '1440p', 'height': 1440, 'bitrate': '8000k'},
{'name': '2160p', 'height': 2160, 'bitrate': '15000k'},
]
# Process each quality
processing_tasks = []
for quality in qualities:
if quality['height'] <= metadata['height']:
task = asyncio.create_task(
self._transcode_to_quality(
original_path, video_id, quality
)
)
processing_tasks.append(task)
# Wait for all transcoding to complete
results = await asyncio.gather(*processing_tasks, return_exceptions=True)
# Generate HLS/DASH manifests
manifests = await self._generate_manifests(video_id, qualities)
# Upload to CDN origin
await self._upload_to_cdn(video_id)
# Update video status
await self.db.execute("""
UPDATE videos
SET processing_status = 'COMPLETED',
processed_at = NOW(),
updated_at = NOW()
WHERE id = %s
""", (video_id,))
# Publish event
await self.kafka.produce('video_processed', {
'event_id': str(uuid.uuid4()),
'event_type': 'VIDEO_PROCESSED',
'video_id': video_id,
'qualities': [q['name'] for q in qualities],
'timestamp': datetime.utcnow().isoformat()
})
# Cleanup temporary files
await self._cleanup_temp_files(video_id)
async def _transcode_to_quality(self, input_path: str, video_id: str,
quality: dict) -> dict:
"""Transcode video to specific quality"""
output_path = f"/tmp/{video_id}/{quality['name']}.mp4"
# FFmpeg command for transcoding
command = [
'ffmpeg',
'-i', input_path,
'-vf', f"scale=-2:{quality['height']}",
'-c:v', 'libx264',
'-preset', 'fast',
'-crf', '23',
'-maxrate', quality['bitrate'],
'-bufsize', f"{int(quality['bitrate'].replace('k', '')) * 2}k",
'-c:a', 'aac',
'-b:a', '128k',
'-f', 'mp4',
output_path
]
# Execute on FFmpeg cluster
result = await self.ffmpeg_cluster.execute(command)
# Upload to S3
await self.s3.upload_file(
output_path,
'video-processed',
f"{video_id}/{quality['name']}.mp4"
)
return {
'quality': quality['name'],
'size': os.path.getsize(output_path),
'duration': result['duration']
}
async def _generate_manifests(self, video_id: str, qualities: List[dict]) -> dict:
"""Generate HLS and DASH manifests for adaptive streaming"""
# HLS manifest
hls_manifest = "#EXTM3U\n#EXT-X-VERSION:3\n"
for quality in qualities:
hls_manifest += (
f"#EXT-X-STREAM-INF:BANDWIDTH={quality['bitrate']},"
f"RESOLUTION={quality['width']}x{quality['height']}\n"
f"{quality['name']}.m3u8\n"
)
# Generate variant playlists
variant_playlists = {}
for quality in qualities:
variant = (
"#EXTM3U\n"
"#EXT-X-VERSION:3\n"
"#EXT-X-TARGETDURATION:10\n"
"#EXT-X-MEDIA-SEQUENCE:0\n"
f"#EXTINF:10.0,\n{quality['name']}_segment_0.ts\n"
f"#EXTINF:10.0,\n{quality['name']}_segment_1.ts\n"
"#EXT-X-ENDLIST\n"
)
variant_playlists[quality['name']] = variant
# DASH manifest (MPD)
dash_manifest = f"""<?xml version="1.0" encoding="UTF-8"?>
<MPD xmlns="urn:mpeg:dash:schema:mpd:2011"
profiles="urn:mpeg:dash:profile:isoff-on-demand:2011"
type="static" mediaPresentationDuration="PT{self.duration}S">
<Period duration="PT{self.duration}S">
"""
for quality in qualities:
dash_manifest += f"""
<AdaptationSet mimeType="video/mp4" codecs="avc1.42c00d">
<Representation id="{quality['name']}"
bandwidth="{quality['bitrate']}"
width="{quality['width']}"
height="{quality['height']}">
<BaseURL>{quality['name']}.mp4</BaseURL>
</Representation>
</AdaptationSet>
"""
dash_manifest += """
</Period>
</MPD>
"""
# Upload manifests
await self.s3.put_object(
bucket='video-processed',
key=f"{video_id}/master.m3u8",
body=hls_manifest.encode()
)
await self.s3.put_object(
bucket='video-processed',
key=f"{video_id}/manifest.mpd",
body=dash_manifest.encode()
)
for quality_name, variant in variant_playlists.items():
await self.s3.put_object(
bucket='video-processed',
key=f"{video_id}/{quality_name}.m3u8",
body=variant.encode()
)
return {
'hls': f"https://cdn.example.com/{video_id}/master.m3u8",
'dash': f"https://cdn.example.com/{video_id}/manifest.mpd"
}
async def _generate_thumbnails(self, video_path: str, video_id: str) -> List[str]:
"""Generate thumbnails at different timestamps"""
thumbnails = []
# Generate at 1%, 25%, 50%, 75%, 99% of duration
timestamps = ['1%', '25%', '50%', '75%', '99%']
for i, timestamp in enumerate(timestamps):
output_path = f"/tmp/{video_id}/thumbnail_{i}.jpg"
command = [
'ffmpeg',
'-i', video_path,
'-ss', timestamp,
'-vframes', '1',
'-vf', 'scale=320:180',
'-q:v', '2',
output_path
]
await self.ffmpeg_cluster.execute(command)
# Upload to S3
thumbnail_url = await self.s3.upload_file(
output_path,
'video-thumbnails',
f"{video_id}/thumbnail_{i}.jpg"
)
thumbnails.append(thumbnail_url)
# Set first thumbnail as default
if i == 0:
await self.db.execute("""
UPDATE videos
SET thumbnail_url = %s
WHERE id = %s
""", (thumbnail_url, video_id))
return thumbnails
3. Streaming Service
python
# services/streaming_service.py
class StreamingService:
"""Video streaming with adaptive bitrate"""
def __init__(self):
self.cdn = CloudFront()
self.redis = RedisCluster()
self.db = PostgreSQL()
self.kafka = KafkaProducer()
self.geoip = GeoIPDatabase()
async def get_stream_url(self, video_id: str, client_info: dict) -> StreamInfo:
"""Get streaming URLs with adaptive bitrate"""
# Check if video exists and is available
video = await self.db.execute("""
SELECT * FROM videos
WHERE id = %s AND status = 'PUBLIC'
""", (video_id,))
if not video:
raise VideoNotFound()
# Get client capabilities
client_caps = self._get_client_capabilities(client_info)
# Get available qualities
qualities = await self._get_available_qualities(video_id)
# Filter based on client capabilities
supported_qualities = self._filter_supported_qualities(
qualities, client_caps
)
# Get CDN URLs
cdn_urls = await self._get_cdn_urls(
video_id, supported_qualities, client_info['ip_address']
)
# Create manifest
manifest = self._create_manifest(
video_id, supported_qualities, cdn_urls
)
# Apply DRM if needed
if video['is_protected']:
drm_info = await self._apply_drm(video_id, manifest)
manifest = drm_info['manifest']
# Log stream request
await self._log_stream_request(
video_id, client_info, supported_qualities
)
return StreamInfo(
video_id=video_id,
manifests=manifest,
qualities=supported_qualities,
subtitles=await self._get_subtitles(video_id),
captions=await self._get_captions(video_id)
)
async def _get_cdn_urls(self, video_id: str, qualities: List[str],
client_ip: str) -> dict:
"""Get optimal CDN URLs based on client location"""
# Determine client location
location = await self.geoip.lookup(client_ip)
# Choose optimal CDN POP
cdn_pop = self._select_cdn_pop(location)
# Generate URLs
urls = {}
for quality in qualities:
if cdn_pop == 'origin':
url = f"https://origin.example.com/{video_id}/{quality}.m3u8"
else:
url = f"https://{cdn_pop}.cdn.example.com/{video_id}/{quality}.m3u8"
urls[quality] = url
return urls
def _select_cdn_pop(self, location: dict) -> str:
"""Select optimal CDN Point of Presence"""
# Simple geographic routing
region = location.get('continent_code', 'EU')
cdn_pops = {
'NA': ['us-east-1', 'us-west-2', 'ca-central-1'],
'EU': ['eu-west-1', 'eu-central-1', 'eu-north-1'],
'AS': ['ap-southeast-1', 'ap-northeast-1', 'ap-south-1'],
'SA': ['sa-east-1'],
'OC': ['ap-southeast-2'],
'AF': ['af-south-1']
}
# Select random POP from region (for load balancing)
pops = cdn_pops.get(region, ['origin'])
return random.choice(pops)
def _create_manifest(self, video_id: str, qualities: List[str],
cdn_urls: dict) -> dict:
"""Create HLS/DASH manifests"""
# Sort qualities (lowest to highest)
quality_order = ['144p', '240p', '360p', '480p',
'720p', '1080p', '1440p', '2160p']
sorted_qualities = [q for q in quality_order if q in qualities]
# HLS Master Playlist
hls_manifest = "#EXTM3U\n#EXT-X-VERSION:6\n"
for quality in sorted_qualities:
hls_manifest += (
f"#EXT-X-STREAM-INF:BANDWIDTH={self._get_bitrate(quality)},"
f"RESOLUTION={self._get_resolution(quality)},"
f"CODECS=\"avc1.42c00d,mp4a.40.2\"\n"
f"{cdn_urls[quality]}\n"
)
# DASH Manifest
dash_manifest = f"""<?xml version="1.0" encoding="UTF-8"?>
<MPD xmlns="urn:mpeg:dash:schema:mpd:2011"
profiles="urn:mpeg:dash:profile:isoff-live:2011"
type="dynamic"
minimumUpdatePeriod="PT2S"
suggestedPresentationDelay="PT5S"
timeShiftBufferDepth="PT30M"
minBufferTime="PT2S">
<Period id="1" start="PT0S">
<AdaptationSet mimeType="video/mp4" codecs="avc1.42c00d">
"""
for quality in sorted_qualities:
dash_manifest += f"""
<Representation id="{quality}"
bandwidth="{self._get_bitrate(quality)}"
width="{self._get_width(quality)}"
height="{self._get_height(quality)}"
frameRate="30">
<SegmentTemplate media="$Bandwidth$/$Number$.m4s"
initialization="$Bandwidth$/init.mp4"
timescale="1000"
startNumber="1"
duration="2000"/>
</Representation>
"""
dash_manifest += """
</AdaptationSet>
</Period>
</MPD>
"""
return {
'hls': hls_manifest,
'dash': dash_manifest,
'urls': cdn_urls
}
async def adaptive_bitrate_logic(self, session_id: str, metrics: dict):
"""Adaptive bitrate switching logic"""
# Get current quality
current_quality = await self.redis.get(
f"stream_session:{session_id}:quality"
)
# Calculate buffer health
buffer_health = metrics['buffer_length'] / metrics['target_buffer']
# Calculate bandwidth estimate
bandwidth = self._estimate_bandwidth(metrics)
# Determine next quality
available_qualities = ['144p', '240p', '360p', '480p',
'720p', '1080p', '1440p', '2160p']
current_index = available_qualities.index(current_quality)
# Simple ABR algorithm (BOLA or similar would be better)
if buffer_health < 0.5:
# Buffer low, downgrade
new_index = max(0, current_index - 2)
elif bandwidth > self._get_bitrate(current_quality) * 1.5:
# Bandwidth good, upgrade
new_index = min(len(available_qualities) - 1, current_index + 1)
elif buffer_health > 1.5:
# Buffer high, consider upgrade
new_index = min(len(available_qualities) - 1, current_index + 1)
else:
# Stay at current quality
new_index = current_index
new_quality = available_qualities[new_index]
# Update session
await self.redis.setex(
f"stream_session:{session_id}:quality",
300, # 5 minutes
new_quality
)
return new_quality
async def handle_live_streaming(self, stream_key: str, input_url: str):
"""Handle live streaming ingestion"""
# Validate stream key
stream = await self.db.execute("""
SELECT * FROM live_streams
WHERE stream_key = %s AND status = 'ACTIVE'
""", (stream_key,))
if not stream:
raise InvalidStreamKey()
# Start ingestion pipeline
ingestion_pipeline = await self._start_ingestion_pipeline(
stream['id'], input_url
)
# Start transcoding
transcoding_pipeline = await self._start_transcoding_pipeline(
stream['id']
)
# Start packaging (HLS/DASH)
packaging_pipeline = await self._start_packaging_pipeline(
stream['id']
)
# Distribute to CDN
distribution_pipeline = await self._start_distribution_pipeline(
stream['id']
)
return {
'stream_id': stream['id'],
'playback_url': f"https://live.example.com/{stream['id']}/master.m3u8",
'ingestion_url': input_url,
'status': 'LIVE'
}
4. Recommendation Engine
python
# services/recommendation_service.py
class RecommendationService:
"""Video recommendations using multiple algorithms"""
def __init__(self):
self.cassandra = Cassandra()
self.redis = RedisCluster()
self.ml_serving = TensorFlowServing()
self.vector_db = Pinecone() # For embeddings
self.kafka = KafkaConsumer()
async def get_recommendations(self, user_id: str, context: dict = None) -> List[Recommendation]:
"""Get personalized video recommendations"""
# Try cache first
cache_key = f"recs:{user_id}:{self._get_context_key(context)}"
cached = await self.redis.get(cache_key)
if cached:
return json.loads(cached)
# Get recommendations from multiple sources
recommendations = await asyncio.gather(
self._get_collaborative_filtering_recs(user_id),
self._get_content_based_recs(user_id),
self._get_trending_recs(context),
self._get_watch_next_recs(user_id),
self._get_similar_to_watched_recs(user_id)
)
# Merge and rank recommendations
merged = self._merge_recommendations(*recommendations)
# Apply diversity and freshness
diversified = await self._apply_diversity_and_freshness(merged)
# Filter out watched videos
filtered = await self._filter_watched_videos(diversified, user_id)
# Cache results
await self.redis.setex(cache_key, 300, json.dumps(filtered[:100]))
return filtered[:50] # Return top 50
async def _get_collaborative_filtering_recs(self, user_id: str) -> List[Recommendation]:
"""Matrix factorization recommendations"""
# Get user embedding
user_embedding = await self.vector_db.fetch(
namespace='user_embeddings',
ids=[user_id]
)
if not user_embedding:
# Cold start: return popular videos
return await self._get_popular_videos()
# Find similar users
similar_users = await self.vector_db.query(
namespace='user_embeddings',
vector=user_embedding[0]['values'],
top_k=50,
include_metadata=True
)
# Get videos liked by similar users
similar_user_ids = [su['id'] for su in similar_users['matches']]
liked_videos = await self.cassandra.execute("""
SELECT video_id, COUNT(*) as score
FROM user_video_interactions
WHERE user_id IN ? AND interaction_type IN ('WATCH', 'LIKE')
GROUP BY video_id
ORDER BY score DESC
LIMIT 200
""", [similar_user_ids])
return [
Recommendation(
video_id=item['video_id'],
score=item['score'] / len(similar_user_ids),
type='collaborative',
reason='Users with similar taste watched'
)
for item in liked_videos
]
async def _get_content_based_recs(self, user_id: str) -> List[Recommendation]:
"""Content-based recommendations using video embeddings"""
# Get user's recent watch history
recent_watches = await self.cassandra.execute("""
SELECT video_id, watch_percentage, timestamp
FROM user_video_interactions
WHERE user_id = %s AND interaction_type = 'WATCH'
ORDER BY timestamp DESC
LIMIT 20
""", (user_id,))
if not recent_watches:
return []
# Get embeddings for watched videos
video_ids = [w['video_id'] for w in recent_watches]
video_embeddings = await self.vector_db.fetch(
namespace='video_embeddings',
ids=video_ids
)
# Calculate weighted average of video embeddings
weights = [min(w['watch_percentage'], 1.0) for w in recent_watches]
user_embedding = self._calculate_weighted_average(
video_embeddings, weights
)
# Find similar videos
similar_videos = await self.vector_db.query(
namespace='video_embeddings',
vector=user_embedding,
top_k=100,
include_metadata=True,
filter={
'status': 'PUBLIC',
'duration': {'$lt': 3600} # Less than 1 hour
}
)
return [
Recommendation(
video_id=match['id'],
score=match['score'],
type='content',
reason='Similar to videos you watched'
)
for match in similar_videos['matches']
]
async def _get_watch_next_recs(self, user_id: str) -> List[Recommendation]:
"""Sequential recommendations (watch next)"""
# Get user's current watching session
current_session = await self.redis.get(f"watch_session:{user_id}")
if not current_session:
return []
session_data = json.loads(current_session)
current_video = session_data.get('current_video')
if not current_video:
return []
# Get videos typically watched after this one
sequence_data = await self.cassandra.execute("""
SELECT next_video_id, COUNT(*) as frequency
FROM video_sequences
WHERE video_id = %s
GROUP BY next_video_id
ORDER BY frequency DESC
LIMIT 50
""", (current_video,))
return [
Recommendation(
video_id=item['next_video_id'],
score=item['frequency'],
type='sequential',
reason='Typically watched next'
)
for item in sequence_data
]
async def _get_similar_to_watched_recs(self, user_id: str) -> List[Recommendation]:
"""Videos similar to user's watched videos"""
# Get top watched categories
watched_categories = await self.cassandra.execute("""
SELECT category, COUNT(*) as watch_count
FROM user_category_watch_stats
WHERE user_id = %s
GROUP BY category
ORDER BY watch_count DESC
LIMIT 5
""", (user_id,))
recommendations = []
for category in watched_categories:
# Get trending videos in category
trending = await self.cassandra.execute("""
SELECT video_id, view_count
FROM category_trending
WHERE category = %s AND time_bucket = %s
ORDER BY view_count DESC
LIMIT 20
""", (category['category'], self._get_current_time_bucket()))
recommendations.extend([
Recommendation(
video_id=item['video_id'],
score=item['view_count'] * category['watch_count'],
type='category',
reason=f'Popular in {category["category"]}'
)
for item in trending
])
return recommendations
def _merge_recommendations(self, *recommendation_lists) -> List[Recommendation]:
"""Merge and re-rank recommendations"""
merged = {}
# Combine scores from different algorithms
for rec_list in recommendation_lists:
for rec in rec_list:
if rec.video_id not in merged:
merged[rec.video_id] = rec
else:
# Weighted combination based on algorithm type
weights = {
'collaborative': 0.35,
'content': 0.25,
'trending': 0.15,
'sequential': 0.15,
'category': 0.10
}
weight = weights.get(rec.type, 0.1)
merged[rec.video_id].score += rec.score * weight
# Sort by score
sorted_recs = sorted(
merged.values(),
key=lambda x: x.score,
reverse=True
)
return sorted_recs
async def _apply_diversity_and_freshness(self, recommendations: List[Recommendation]) -> List[Recommendation]:
"""Apply diversity and freshness to recommendations"""
diversified = []
categories_seen = set()
creators_seen = set()
for rec in recommendations:
# Get video metadata
video = await self._get_video_metadata(rec.video_id)
# Check diversity
if (video['category'] in categories_seen and
video['creator_id'] in creators_seen):
continue # Skip if too similar
# Apply freshness boost
freshness_boost = self._calculate_freshness_boost(video['created_at'])
rec.score *= freshness_boost
diversified.append(rec)
# Update seen sets
categories_seen.add(video['category'])
creators_seen.add(video['creator_id'])
if len(diversified) >= 100:
break
return diversified
5. Live Streaming Service
python
# services/live_streaming_service.py
class LiveStreamingService:
"""Real-time live streaming pipeline"""
def __init__(self):
self.media_server = MediaServerCluster()
self.transcoding = TranscodingCluster()
self.cdn = CDNManager()
self.redis = RedisCluster()
self.db = PostgreSQL()
self.kafka = KafkaProducer()
async def start_live_stream(self, user_id: str, stream_config: dict) -> LiveStream:
"""Start a new live stream"""
# Generate stream ID and key
stream_id = self._generate_stream_id()
stream_key = self._generate_stream_key()
# Create stream record
await self.db.execute("""
INSERT INTO live_streams (
id, user_id, title, description,
stream_key, status, created_at
) VALUES (%s, %s, %s, %s, %s, %s, NOW())
""", (
stream_id, user_id, stream_config['title'],
stream_config['description'], stream_key, 'CREATED'
))
# Get ingest URL
ingest_url = await self._get_ingest_url(stream_id)
# Start media server for this stream
await self.media_server.create_stream(
stream_id=stream_id,
config={
'protocols': ['rtmp', 'srt', 'rtsp'],
'codecs': ['h264', 'h265', 'vp9'],
'max_bitrate': stream_config.get('max_bitrate', 10000000),
'max_resolution': stream_config.get('max_resolution', '1080p')
}
)
# Create transcoding profiles
transcoding_profiles = await self._create_transcoding_profiles(
stream_id, stream_config
)
return LiveStream(
stream_id=stream_id,
stream_key=stream_key,
ingest_url=ingest_url,
playback_urls=self._generate_playback_urls(stream_id),
status='READY',
config=stream_config
)
async def process_live_stream(self, stream_id: str):
"""Process incoming live stream"""
# Get stream from media server
stream = await self.media_server.get_stream(stream_id)
# Validate stream health
if not await self._validate_stream_health(stream):
await self._update_stream_status(stream_id, 'ERROR')
return
# Update stream status
await self._update_stream_status(stream_id, 'LIVE')
# Start transcoding pipeline
transcoding_pipeline = await self.transcoding.start_pipeline(
input_url=stream['playout_url'],
stream_id=stream_id,
profiles=[
{'name': 'source', 'copy': True},
{'name': '1080p', 'height': 1080, 'bitrate': '5000k'},
{'name': '720p', 'height': 720, 'bitrate': '2500k'},
{'name': '480p', 'height': 480, 'bitrate': '1200k'},
{'name': '360p', 'height': 360, 'bitrate': '800k'},
{'name': '240p', 'height': 240, 'bitrate': '400k'},
]
)
# Start packaging (HLS/DASH)
packaging_pipeline = await self._start_packaging_pipeline(
stream_id, transcoding_pipeline['outputs']
)
# Distribute to CDN
distribution = await self.cdn.distribute_live_stream(
stream_id, packaging_pipeline['manifests']
)
# Start DVR (Digital Video Recording)
if self._should_record_stream(stream_id):
await self._start_recording(stream_id, stream['playout_url'])
# Start real-time analytics
await self._start_real_time_analytics(stream_id)
# Monitor stream health
asyncio.create_task(
self._monitor_stream_health(stream_id)
)
async def _start_packaging_pipeline(self, stream_id: str, transcoded_streams: dict):
"""Package streams into HLS/DASH formats"""
packaging_tasks = []
for quality, stream_url in transcoded_streams.items():
# HLS packaging
hls_task = asyncio.create_task(
self._package_hls(stream_id, quality, stream_url)
)
packaging_tasks.append(hls_task)
# DASH packaging
dash_task = asyncio.create_task(
self._package_dash(stream_id, quality, stream_url)
)
packaging_tasks.append(dash_task)
# Wait for all packaging to complete
results = await asyncio.gather(*packaging_tasks)
# Generate master manifests
hls_manifest = self._generate_hls_master_manifest(
stream_id, results
)
dash_manifest = self._generate_dash_master_manifest(
stream_id, results
)
# Upload manifests
await self.cdn.upload_manifest(
stream_id, 'hls', hls_manifest
)
await self.cdn.upload_manifest(
stream_id, 'dash', dash_manifest
)
return {
'manifests': {
'hls': f"https://live.example.com/{stream_id}/master.m3u8",
'dash': f"https://live.example.com/{stream_id}/manifest.mpd"
},
'qualities': list(transcoded_streams.keys())
}
async def _package_hls(self, stream_id: str, quality: str, input_url: str) -> dict:
"""Package stream into HLS format"""
output_dir = f"/tmp/live/{stream_id}/{quality}"
os.makedirs(output_dir, exist_ok=True)
# FFmpeg command for HLS packaging
command = [
'ffmpeg',
'-i', input_url,
'-c', 'copy',
'-f', 'hls',
'-hls_time', '2',
'-hls_list_size', '10',
'-hls_flags', 'delete_segments',
'-hls_segment_filename', f'{output_dir}/segment_%03d.ts',
f'{output_dir}/playlist.m3u8'
]
# Start packaging process
process = await asyncio.create_subprocess_exec(*command)
# Monitor process
asyncio.create_task(
self._monitor_packaging_process(stream_id, quality, process)
)
return {
'quality': quality,
'playlist_url': f"{output_dir}/playlist.m3u8",
'segment_pattern': f"{output_dir}/segment_%03d.ts",
'process': process
}
async def handle_low_latency_streaming(self, stream_id: str):
"""Handle low-latency streaming (WebRTC, LL-HLS)"""
# WebRTC ingestion
webrtc_config = await self._setup_webrtc_ingestion(stream_id)
# Low-latency HLS (LL-HLS)
ll_hls_config = await self._setup_ll_hls(stream_id)
# CMAF for low latency DASH
cmaf_config = await self._setup_cmaf(stream_id)
return {
'webrtc': {
'ingest_url': webrtc_config['ingest_url'],
'playback_url': webrtc_config['playback_url'],
'latency': '500ms'
},
'll_hls': {
'playback_url': ll_hls_config['playback_url'],
'latency': '3s'
},
'cmaf': {
'playback_url': cmaf_config['playback_url'],
'latency': '3s'
}
}
async def _monitor_stream_health(self, stream_id: str):
"""Monitor live stream health"""
while True:
try:
# Check stream health
health = await self._check_stream_health(stream_id)
if health['status'] == 'HEALTHY':
# Update metrics
await self._update_stream_metrics(stream_id, health['metrics'])
# Check for quality issues
if health['metrics']['dropped_frames'] > 10:
await self._handle_quality_issue(stream_id)
elif health['status'] == 'UNHEALTHY':
# Attempt recovery
await self._attempt_stream_recovery(stream_id)
# If recovery fails, end stream
if not await self._check_stream_health(stream_id):
await self.end_live_stream(stream_id, 'ERROR')
break
# Sleep for monitoring interval
await asyncio.sleep(5)
except Exception as e:
logger.error(f"Error monitoring stream {stream_id}: {e}")
await asyncio.sleep(10)
async def end_live_stream(self, stream_id: str, reason: str = 'ENDED'):
"""End a live stream"""
# Update stream status
await self._update_stream_status(stream_id, 'ENDING')
# Stop ingestion
await self.media_server.stop_stream(stream_id)
# Stop transcoding
await self.transcoding.stop_pipeline(stream_id)
# Stop packaging
await self._stop_packaging_pipeline(stream_id)
# Finalize recording if exists
if await self._has_recording(stream_id):
recording = await self._finalize_recording(stream_id)
# Create VOD from recording
vod_id = await self._create_vod_from_recording(recording)
# Update stream with VOD link
await self.db.execute("""
UPDATE live_streams
SET vod_id = %s, ended_at = NOW()
WHERE id = %s
""", (vod_id, stream_id))
# Update final status
await self._update_stream_status(stream_id, 'ENDED')
# Clean up resources
await self._cleanup_stream_resources(stream_id)
# Publish event
await self.kafka.produce('live_stream_events', {
'event_id': str(uuid.uuid4()),
'event_type': 'LIVE_STREAM_ENDED',
'stream_id': stream_id,
'reason': reason,
'timestamp': datetime.utcnow().isoformat()
})
6. Content Delivery Network (CDN)
python
# services/cdn_service.py
class CDNService:
"""Global content delivery network management"""
def __init__(self):
self.cloudfront = CloudFront()
self.akamai = Akamai()
self.fastly = Fastly()
self.s3 = S3()
self.redis = RedisCluster()
async def distribute_video(self, video_id: str, video_files: dict):
"""Distribute video files to CDN"""
# Upload to origin (S3)
await self._upload_to_origin(video_id, video_files)
# Invalidate CDN cache
await self._invalidate_cdn_cache(video_id)
# Pre-warm popular regions
await self._prewarm_popular_regions(video_id)
# Set up origin shield
await self._setup_origin_shield(video_id)
return {
'cdn_urls': self._generate_cdn_urls(video_id),
'regions': self._get_covered_regions(),
'ttl': 86400 # 24 hours
}
async def _upload_to_origin(self, video_id: str, video_files: dict):
"""Upload video files to origin storage"""
# Upload each quality
upload_tasks = []
for quality, file_path in video_files.items():
task = asyncio.create_task(
self.s3.upload_file(
file_path,
'video-origin',
f"{video_id}/{quality}"
)
)
upload_tasks.append(task)
# Upload manifests
manifest_tasks = []
for manifest_type in ['hls', 'dash']:
task = asyncio.create_task(
self.s3.upload_manifest(
video_id, manifest_type
)
)
manifest_tasks.append(task)
# Wait for all uploads
await asyncio.gather(*upload_tasks, *manifest_tasks)
# Set cache headers
await self.s3.set_cache_headers(
bucket='video-origin',
prefix=video_id,
headers={
'Cache-Control': 'public, max-age=31536000', # 1 year
'Content-Type': 'video/mp4'
}
)
async def _invalidate_cdn_cache(self, video_id: str):
"""Invalidate CDN cache for video"""
# CloudFront invalidation
cf_invalidation = await self.cloudfront.create_invalidation(
distribution_id='E1A2B3C4D5E6F7',
paths=[f'/videos/{video_id}/*']
)
# Akamai cache purge
akamai_purge = await self.akamai.purge(
objects=[f'https://video.example.com/{video_id}/*'],
action='invalidate'
)
# Fastly purge
fastly_purge = await self.fastly.purge(
service_id='service123',
keys=[f'video:{video_id}']
)
return {
'cloudfront': cf_invalidation['Id'],
'akamai': akamai_purge['purgeId'],
'fastly': fastly_purge['id']
}
async def _prewarm_popular_regions(self, video_id: str):
"""Pre-warm cache in popular regions"""
popular_regions = [
'us-east-1', 'us-west-2', # North America
'eu-west-1', 'eu-central-1', # Europe
'ap-southeast-1', 'ap-northeast-1', # Asia
]
# Generate pre-warm URLs
urls = []
for quality in ['144p', '360p', '720p', '1080p']:
urls.append(f"https://video.example.com/{video_id}/{quality}.m3u8")
urls.append(f"https://video.example.com/{video_id}/{quality}_segment_0.ts")
# Pre-warm each region
for region in popular_regions:
await self.cloudfront.prewarm(
distribution_id='E1A2B3C4D5E6F7',
region=region,
urls=urls[:10] # Limit to first 10 URLs
)
async def _setup_origin_shield(self, video_id: str):
"""Setup origin shield for popular content"""
# Check if video is expected to be popular
is_popular = await self._is_expected_popular(video_id)
if is_popular:
# Setup regional caches as origin shields
await self.cloudfront.update_distribution(
distribution_id='E1A2B3C4D5E6F7',
origin_shield={
'Enabled': True,
'OriginShieldRegion': 'us-east-1'
}
)
async def optimize_delivery(self, request, video_id: str):
"""Optimize delivery based on client and network conditions"""
# Get client capabilities
client_info = self._get_client_info(request)
# Get network conditions
network_info = await self._get_network_info(request)
# Select optimal CDN
cdn = self._select_optimal_cdn(client_info, network_info)
# Select optimal protocol
protocol = self._select_optimal_protocol(client_info, network_info)
# Select optimal bitrate
bitrate = self._select_optimal_bitrate(network_info)
# Generate optimized URL
optimized_url = self._generate_optimized_url(
video_id, cdn, protocol, bitrate
)
return {
'url': optimized_url,
'cdn': cdn,
'protocol': protocol,
'bitrate': bitrate,
'optimizations': {
'compression': 'br' if client_info.get('supports_brotli') else 'gzip',
'image_format': 'webp' if client_info.get('supports_webp') else 'jpeg',
'video_codec': 'h265' if client_info.get('supports_h265') else 'h264'
}
}
async def handle_live_cdn(self, stream_id: str, manifests: dict):
"""Handle live streaming CDN distribution"""
# Setup live streaming origin
origin = await self._setup_live_origin(stream_id)
# Setup edge locations
edges = await self._setup_edge_locations(stream_id)
# Setup real-time manifest updates
manifest_updater = await self._setup_manifest_updater(stream_id)
# Setup low-latency delivery
low_latency = await self._setup_low_latency_delivery(stream_id)
return {
'origin': origin['url'],
'edges': edges,
'playback_urls': {
'hls': f"https://live.example.com/{stream_id}/master.m3u8",
'dash': f"https://live.example.com/{stream_id}/manifest.mpd",
'll_hls': f"https://live.example.com/{stream_id}/llhls.m3u8"
},
'latency': '3s'
}
Database Schema Design
1. PostgreSQL Schema (Metadata, Users, Analytics)
sql
-- Videos table (partitioned by created_at)
CREATE TABLE videos (
id VARCHAR(50) PRIMARY KEY,
user_id BIGINT NOT NULL,
title VARCHAR(500) NOT NULL,
description TEXT,
duration INT NOT NULL, -- in seconds
width INT,
height INT,
framerate DECIMAL(5,2),
codec VARCHAR(50),
bitrate INT,
file_size BIGINT,
thumbnail_url TEXT,
category VARCHAR(100),
tags TEXT[],
privacy_status VARCHAR(20) DEFAULT 'PUBLIC', -- PUBLIC, PRIVATE, UNLISTED
processing_status VARCHAR(20) DEFAULT 'PENDING', -- PENDING, PROCESSING, COMPLETED, FAILED
view_count BIGINT DEFAULT 0,
like_count INT DEFAULT 0,
dislike_count INT DEFAULT 0,
comment_count INT DEFAULT 0,
is_live BOOLEAN DEFAULT false,
is_premium BOOLEAN DEFAULT false,
is_360 BOOLEAN DEFAULT false,
is_hdr BOOLEAN DEFAULT false,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
published_at TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
INDEX idx_user_id (user_id),
INDEX idx_category (category),
INDEX idx_created_at (created_at),
INDEX idx_view_count (view_count),
INDEX idx_like_count (like_count),
FULLTEXT INDEX idx_title_description (title, description)
) PARTITION BY RANGE (created_at);
-- Video qualities
CREATE TABLE video_qualities (
video_id VARCHAR(50) NOT NULL REFERENCES videos(id),
quality VARCHAR(20) NOT NULL, -- 144p, 240p, 360p, etc.
width INT NOT NULL,
height INT NOT NULL,
bitrate INT NOT NULL,
file_size BIGINT NOT NULL,
url TEXT NOT NULL,
manifest_url TEXT,
codec VARCHAR(50),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (video_id, quality)
);
-- Users
CREATE TABLE users (
id BIGSERIAL PRIMARY KEY,
username VARCHAR(100) UNIQUE NOT NULL,
email VARCHAR(255) UNIQUE NOT NULL,
password_hash VARCHAR(255) NOT NULL,
channel_name VARCHAR(200),
channel_description TEXT,
avatar_url TEXT,
banner_url TEXT,
subscriber_count INT DEFAULT 0,
video_count INT DEFAULT 0,
total_views BIGINT DEFAULT 0,
country VARCHAR(2),
language VARCHAR(10),
is_verified BOOLEAN DEFAULT false,
is_partner BOOLEAN DEFAULT false,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
INDEX idx_username (username),
INDEX idx_email (email),
INDEX idx_created_at (created_at)
);
-- Watch history
CREATE TABLE watch_history (
id BIGSERIAL PRIMARY KEY,
user_id BIGINT NOT NULL REFERENCES users(id),
video_id VARCHAR(50) NOT NULL REFERENCES videos(id),
watch_percentage DECIMAL(5,2) NOT NULL, -- 0 to 100
watch_duration INT NOT NULL, -- in seconds
quality VARCHAR(20),
device_type VARCHAR(50),
location VARCHAR(100),
watched_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE (user_id, video_id, watched_at),
INDEX idx_user_video (user_id, video_id),
INDEX idx_watched_at (watched_at)
) PARTITION BY RANGE (watched_at);
-- Subscriptions
CREATE TABLE subscriptions (
id BIGSERIAL PRIMARY KEY,
subscriber_id BIGINT NOT NULL REFERENCES users(id),
channel_id BIGINT NOT NULL REFERENCES users(id),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE (subscriber_id, channel_id),
INDEX idx_subscriber (subscriber_id),
INDEX idx_channel (channel_id)
);
-- Comments
CREATE TABLE comments (
id BIGSERIAL PRIMARY KEY,
video_id VARCHAR(50) NOT NULL REFERENCES videos(id),
user_id BIGINT NOT NULL REFERENCES users(id),
parent_id BIGINT REFERENCES comments(id), -- for replies
comment_text TEXT NOT NULL,
like_count INT DEFAULT 0,
is_pinned BOOLEAN DEFAULT false,
is_edited BOOLEAN DEFAULT false,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
INDEX idx_video (video_id),
INDEX idx_user (user_id),
INDEX idx_parent (parent_id),
INDEX idx_created_at (created_at)
);
-- Playlists
CREATE TABLE playlists (
id VARCHAR(50) PRIMARY KEY,
user_id BIGINT NOT NULL REFERENCES users(id),
title VARCHAR(200) NOT NULL,
description TEXT,
privacy_status VARCHAR(20) DEFAULT 'PUBLIC', -- PUBLIC, PRIVATE, UNLISTED
video_count INT DEFAULT 0,
total_duration INT DEFAULT 0, -- in seconds
thumbnail_video_id VARCHAR(50) REFERENCES videos(id),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
INDEX idx_user_id (user_id)
);
-- Playlist items
CREATE TABLE playlist_items (
playlist_id VARCHAR(50) NOT NULL REFERENCES playlists(id),
video_id VARCHAR(50) NOT NULL REFERENCES videos(id),
position INT NOT NULL,
added_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (playlist_id, video_id),
INDEX idx_playlist_position (playlist_id, position)
);
2. Cassandra Schema (Time-series, User Interactions)
cql
-- User video interactions (for recommendations)
CREATE TABLE user_video_interactions (
user_id TEXT,
video_id TEXT,
interaction_type TEXT, -- WATCH, LIKE, DISLIKE, SHARE, COMMENT
watch_percentage DECIMAL,
watch_duration INT,
device_type TEXT,
location TEXT,
timestamp TIMESTAMP,
PRIMARY KEY ((user_id), video_id, interaction_type, timestamp)
) WITH CLUSTERING ORDER BY (video_id ASC, interaction_type ASC, timestamp DESC);
-- Video view counters (for trending)
CREATE TABLE video_view_counters (
video_id TEXT,
time_bucket TEXT, -- hour:2024-01-15-10, day:2024-01-15
view_count COUNTER,
watch_seconds COUNTER,
PRIMARY KEY ((video_id), time_bucket)
);
-- Category trending
CREATE TABLE category_trending (
category TEXT,
time_bucket TEXT, -- hour:2024-01-15-10
video_id TEXT,
view_count COUNTER,
PRIMARY KEY ((category, time_bucket), video_id)
) WITH CLUSTERING ORDER BY (video_id ASC);
-- Live stream viewers
CREATE TABLE live_stream_viewers (
stream_id TEXT,
user_id TEXT,
joined_at TIMESTAMP,
last_heartbeat TIMESTAMP,
device_type TEXT,
location TEXT,
PRIMARY KEY ((stream_id), user_id)
);
-- Chat messages for live streams
CREATE TABLE live_stream_chat (
stream_id TEXT,
message_id TIMEUUID,
user_id TEXT,
username TEXT,
message TEXT,
is_moderator BOOLEAN,
is_subscriber BOOLEAN,
timestamp TIMESTAMP,
PRIMARY KEY ((stream_id), message_id)
) WITH CLUSTERING ORDER BY (message_id DESC);
-- User session data
CREATE TABLE user_sessions (
session_id TEXT PRIMARY KEY,
user_id TEXT,
device_id TEXT,
device_type TEXT, -- MOBILE, DESKTOP, TV
app_version TEXT,
os_version TEXT,
screen_resolution TEXT,
supported_codecs SET<TEXT>,
max_bitrate INT,
location TEXT,
started_at TIMESTAMP,
last_active_at TIMESTAMP,
session_data MAP<TEXT, TEXT>
);
3. MongoDB Schema (Unstructured Data, Comments)
javascript
// Video metadata with embedded thumbnails
{
"_id": "dQw4w9WgXcQ",
"title": "Never Gonna Give You Up",
"description": "Official video for Rick Astley - Never Gonna Give You Up",
"channel_id": "UCuAXFkgsw1L7xaCfnd5JJOw",
"channel_name": "Rick Astley",
"duration": 213,
"published_at": ISODate("2009-10-25T06:57:33Z"),
"thumbnails": {
"default": {
"url": "https://i.ytimg.com/vi/dQw4w9WgXcQ/default.jpg",
"width": 120,
"height": 90
},
"medium": {
"url": "https://i.ytimg.com/vi/dQw4w9WgXcQ/mqdefault.jpg",
"width": 320,
"height": 180
},
"high": {
"url": "https://i.ytimg.com/vi/dQw4w9WgXcQ/hqdefault.jpg",
"width": 480,
"height": 360
},
"standard": {
"url": "https://i.ytimg.com/vi/dQw4w9WgXcQ/sddefault.jpg",
"width": 640,
"height": 480
},
"maxres": {
"url": "https://i.ytimg.com/vi/dQw4w9WgXcQ/maxresdefault.jpg",
"width": 1280,
"height": 720
}
},
"tags": ["rick astley", "never gonna give you up", "80s", "music video"],
"category_id": "10", // Music
"statistics": {
"view_count": 1345678901,
"like_count": 15678901,
"dislike_count": 234567,
"favorite_count": 0,
"comment_count": 4567890
},
"content_details": {
"dimension": "2d",
"definition": "hd",
"caption": true,
"licensed_content": true,
"region_restriction": {
"allowed": ["US", "CA", "GB", "AU"],
"blocked": ["DE", "FR"]
}
},
"status": {
"upload_status": "processed",
"privacy_status": "public",
"license": "youtube",
"embeddable": true,
"public_stats_viewable": true
},
"processing_details": {
"processing_status": "succeeded",
"processing_progress": {
"parts_total": 1,
"parts_processed": 1,
"time_left_ms": 0
}
},
"live_streaming_details": {
"actual_start_time": ISODate("2024-01-15T10:00:00Z"),
"actual_end_time": ISODate("2024-01-15T11:30:00Z"),
"scheduled_start_time": ISODate("2024-01-15T10:00:00Z"),
"concurrent_viewers": 125000,
"active_live_chat_id": "abc123"
}
}
// Comments with replies (nested structure)
{
"_id": "Ugz-7zMTR4e8qE3sJjh4AaABAg",
"video_id": "dQw4w9WgXcQ",
"author": {
"id": "UCuAXFkgsw1L7xaCfnd5JJOw",
"name": "Rick Astley",
"avatar": "https://yt3.ggpht.com/a/AGF-l78jvYV3FqJ6ZQ7ZLQ"
},
"text": "Thanks for watching!",
"like_count": 156789,
"published_at": ISODate("2009-10-25T07:00:00Z"),
"updated_at": ISODate("2009-10-25T07:00:00Z"),
"total_reply_count": 12345,
"replies": [
{
"id": "Ugz-7zMTR4e8qE3sJjh4AaABAg.123",
"author": {
"id": "UC1234567890",
"name": "Fan User",
"avatar": "https://yt3.ggpht.com/a/default-user"
},
"text": "Love this song!",
"like_count": 1234,
"published_at": ISODate("2009-10-25T07:05:00Z")
}
],
"moderation": {
"held_for_review": false,
"is_public": true
}
}
// User profiles with preferences
{
"_id": "UCuAXFkgsw1L7xaCfnd5JJOw",
"username": "RickAstley",
"display_name": "Rick Astley",
"description": "Official Rick Astley YouTube channel",
"country": "GB",
"joined_date": ISODate("2006-03-01T00:00:00Z"),
"stats": {
"subscribers": 15678901,
"views": 3456789012,
"videos": 123
},
"branding": {
"banner": {
"url": "https://yt3.ggpht.com/banner.jpg",
"width": 2560,
"height": 1440
},
"avatar": {
"url": "https://yt3.ggpht.com/avatar.jpg",
"width": 800,
"height": 800
}
},
"preferences": {
"language": "en",
"country": "GB",
"notifications": {
"subscriptions": true,
"recommended": true,
"comments": false
},
"playback": {
"default_quality": "auto",
"autoplay": true,
"captions": "en"
},
"privacy": {
"watch_history": true,
"search_history": true,
"liked_videos": "public"
}
},
"monetization": {
"is_monetized": true,
"revenue_split": 0.55,
"ad_formats": ["pre-roll", "mid-roll", "display"]
}
}
4. Elasticsearch Schema (Search & Discovery)
json
{
"settings": {
"number_of_shards": 20,
"number_of_replicas": 2,
"analysis": {
"analyzer": {
"video_analyzer": {
"type": "custom",
"tokenizer": "standard",
"filter": ["lowercase", "asciifolding", "english_stemmer"]
},
"autocomplete_analyzer": {
"type": "custom",
"tokenizer": "standard",
"filter": ["lowercase", "autocomplete_filter"]
},
"autocomplete_search": {
"type": "custom",
"tokenizer": "standard",
"filter": ["lowercase"]
}
},
"filter": {
"autocomplete_filter": {
"type": "edge_ngram",
"min_gram": 2,
"max_gram": 20
},
"english_stemmer": {
"type": "stemmer",
"language": "english"
}
}
}
},
"mappings": {
"properties": {
"video_id": {"type": "keyword"},
"title": {
"type": "text",
"analyzer": "video_analyzer",
"fields": {
"autocomplete": {
"type": "text",
"analyzer": "autocomplete_analyzer",
"search_analyzer": "autocomplete_search"
},
"keyword": {"type": "keyword"}
}
},
"description": {"type": "text", "analyzer": "video_analyzer"},
"channel_id": {"type": "keyword"},
"channel_name": {"type": "keyword"},
"duration": {"type": "integer"},
"published_at": {"type": "date"},
"view_count": {"type": "long"},
"like_count": {"type": "long"},
"comment_count": {"type": "long"},
"category": {"type": "keyword"},
"tags": {"type": "keyword"},
"content_type": {"type": "keyword"}, // video, short, live, podcast
"is_live": {"type": "boolean"},
"is_premium": {"type": "boolean"},
"resolution": {"type": "keyword"}, // sd, hd, fhd, uhd
"language": {"type": "keyword"},
"captions": {"type": "boolean"},
"license": {"type": "keyword"},
"embeddable": {"type": "boolean"},
"age_restricted": {"type": "boolean"},
"has_thumbnail": {"type": "boolean"},
"has_captions": {"type": "boolean"},
"engagement_score": {"type": "float"}, // Calculated score for ranking
"freshness_score": {"type": "float"}, // Based on publication date
"quality_score": {"type": "float"}, // Based on resolution, bitrate, etc.
"location": {
"type": "geo_point"
},
"suggest": {
"type": "completion",
"contexts": [
{
"name": "category",
"type": "category"
}
]
}
}
}
}
Video Processing Pipeline Architecture
FFmpeg Processing Cluster
python
# processing/ffmpeg_cluster.py
class FFMpegCluster:
"""Distributed FFmpeg processing cluster"""
def __init__(self):
self.workers = KubernetesCluster()
self.job_queue = RabbitMQ()
self.result_store = Redis()
self.monitoring = Prometheus()
async def process_video(self, video_id: str, input_path: str,
processing_profile: dict) -> dict:
"""Process video through distributed FFmpeg cluster"""
# Create processing jobs
jobs = self._create_processing_jobs(
video_id, input_path, processing_profile
)
# Submit jobs to queue
job_ids = []
for job in jobs:
job_id = await self.job_queue.submit(job)
job_ids.append(job_id)
# Monitor job progress
asyncio.create_task(
self._monitor_job(job_id, video_id)
)
# Wait for all jobs to complete
results = await asyncio.gather(*[
self._wait_for_job(job_id) for job_id in job_ids
])
# Combine results
combined_results = self._combine_results(results)
# Clean up temporary files
await self._cleanup_temp_files(video_id)
return combined_results
def _create_processing_jobs(self, video_id: str, input_path: str,
profile: dict) -> List[dict]:
"""Create processing jobs for different operations"""
jobs = []
# Thumbnail generation jobs
thumbnail_jobs = self._create_thumbnail_jobs(
video_id, input_path, profile.get('thumbnails', {})
)
jobs.extend(thumbnail_jobs)
# Transcoding jobs for different qualities
transcoding_jobs = self._create_transcoding_jobs(
video_id, input_path, profile.get('qualities', [])
)
jobs.extend(transcoding_jobs)
# Audio extraction jobs
audio_jobs = self._create_audio_jobs(video_id, input_path)
jobs.extend(audio_jobs)
# Subtitle extraction jobs
subtitle_jobs = self._create_subtitle_jobs(video_id, input_path)
jobs.extend(subtitle_jobs)
# Metadata extraction job
metadata_job = self._create_metadata_job(video_id, input_path)
jobs.append(metadata_job)
return jobs
def _create_transcoding_jobs(self, video_id: str, input_path: str,
qualities: List[dict]) -> List[dict]:
"""Create transcoding jobs for different qualities"""
jobs = []
for quality in qualities:
job = {
'type': 'transcode',
'video_id': video_id,
'input_path': input_path,
'output_format': quality.get('format', 'mp4'),
'quality': quality['name'],
'height': quality['height'],
'bitrate': quality['bitrate'],
'codec': quality.get('codec', 'h264'),
'preset': quality.get('preset', 'fast'),
'crf': quality.get('crf', 23),
'audio_bitrate': quality.get('audio_bitrate', '128k'),
'output_path': f"/tmp/{video_id}/{quality['name']}.mp4",
'priority': self._get_priority(quality['name'])
}
jobs.append(job)
return jobs
def _create_thumbnail_jobs(self, video_id: str, input_path: str,
config: dict) -> List[dict]:
"""Create thumbnail generation jobs"""
jobs = []
# Generate thumbnails at different timestamps
timestamps = config.get('timestamps', ['1%', '25%', '50%', '75%', '99%'])
sizes = config.get('sizes', [
{'name': 'default', 'width': 120, 'height': 90},
{'name': 'medium', 'width': 320, 'height': 180},
{'name': 'high', 'width': 480, 'height': 360},
{'name': 'standard', 'width': 640, 'height': 480},
{'name': 'maxres', 'width': 1280, 'height': 720}
])
for i, timestamp in enumerate(timestamps):
for size in sizes:
job = {
'type': 'thumbnail',
'video_id': video_id,
'input_path': input_path,
'timestamp': timestamp,
'width': size['width'],
'height': size['height'],
'quality': 85,
'output_path': f"/tmp/{video_id}/thumbnails/{size['name']}_{i}.jpg",
'priority': 'low'
}
jobs.append(job)
return jobs
async def _execute_job_on_worker(self, job: dict) -> dict:
"""Execute job on FFmpeg worker"""
# Select worker based on job requirements
worker = await self._select_worker(job)
# Execute FFmpeg command
command = self._build_ffmpeg_command(job)
try:
# Execute on worker
result = await worker.execute(command, timeout=3600)
# Upload result to storage
output_url = await self._upload_result(job, result['output_path'])
return {
'job_id': job['job_id'],
'status': 'completed',
'output_url': output_url,
'processing_time': result['processing_time'],
'output_size': result['output_size']
}
except Exception as e:
logger.error(f"Job failed: {job['job_id']}, error: {e}")
# Retry logic
if job.get('retry_count', 0) < 3:
job['retry_count'] = job.get('retry_count', 0) + 1
await self.job_queue.retry(job)
else:
return {
'job_id': job['job_id'],
'status': 'failed',
'error': str(e)
}
def _build_ffmpeg_command(self, job: dict) -> List[str]:
"""Build FFmpeg command based on job type"""
if job['type'] == 'transcode':
return self._build_transcode_command(job)
elif job['type'] == 'thumbnail':
return self._build_thumbnail_command(job)
elif job['type'] == 'metadata':
return self._build_metadata_command(job)
elif job['type'] == 'audio':
return self._build_audio_command(job)
elif job['type'] == 'subtitle':
return self._build_subtitle_command(job)
else:
raise ValueError(f"Unknown job type: {job['type']}")
def _build_transcode_command(self, job: dict) -> List[str]:
"""Build FFmpeg command for transcoding"""
cmd = [
'ffmpeg',
'-i', job['input_path'],
'-vf', f"scale=-2:{job['height']}",
'-c:v', job.get('codec', 'libx264'),
'-preset', job.get('preset', 'fast'),
'-crf', str(job.get('crf', 23)),
'-maxrate', job['bitrate'],
'-bufsize', f"{int(job['bitrate'].replace('k', '')) * 2}k",
'-c:a', 'aac',
'-b:a', job.get('audio_bitrate', '128k'),
'-movflags', '+faststart',
'-y', # Overwrite output
job['output_path']
]
# Add additional parameters
if job.get('codec') == 'h265':
cmd.extend(['-tag:v', 'hvc1'])
if job.get('two_pass', False):
# First pass
cmd1 = cmd.copy()
cmd1.extend(['-pass', '1', '-f', 'null', '/dev/null'])
# Second pass
cmd2 = cmd.copy()
cmd2.extend(['-pass', '2'])
return [cmd1, cmd2]
return cmd
async def _monitor_cluster_health(self):
"""Monitor FFmpeg cluster health"""
while True:
try:
# Check worker health
workers = await self.workers.list()
healthy_workers = []
for worker in workers:
health = await self._check_worker_health(worker)
if health['status'] == 'healthy':
healthy_workers.append(worker)
else:
# Handle unhealthy worker
await self._handle_unhealthy_worker(worker)
# Adjust cluster size based on queue
queue_size = await self.job_queue.size()
desired_workers = self._calculate_desired_workers(queue_size)
if len(healthy_workers) < desired_workers:
# Scale up
await self._scale_up(desired_workers - len(healthy_workers))
elif len(healthy_workers) > desired_workers * 1.5:
# Scale down
await self._scale_down(len(healthy_workers) - desired_workers)
# Update metrics
await self.monitoring.update_metrics({
'workers_total': len(workers),
'workers_healthy': len(healthy_workers),
'queue_size': queue_size,
'jobs_processed': await self._get_jobs_processed()
})
await asyncio.sleep(60) # Check every minute
except Exception as e:
logger.error(f"Error monitoring cluster: {e}")
await asyncio.sleep(30)
Adaptive Bitrate Streaming (ABR) Algorithms
1. BOLA (Buffer Occupancy based Lyapunov Algorithm)
python
# streaming/abr_algorithms.py
class BOLAAlgorithm:
"""BOLA ABR algorithm implementation"""
def __init__(self, config: dict = None):
self.config = config or {}
self.gp = self.config.get('gp', 5) # Penalty factor
self.Vp = self.config.get('Vp', 0.14) # Control parameter
async def select_quality(self, session_state: dict,
available_qualities: List[dict]) -> str:
"""Select optimal quality using BOLA algorithm"""
# Current buffer level
buffer_level = session_state.get('buffer_level', 0)
# Available bitrates
bitrates = [q['bitrate'] for q in available_qualities]
sizes = [q['size'] for q in available_qualities]
qualities = [q['name'] for q in available_qualities]
# Calculate utility for each quality
utilities = []
for bitrate, size in zip(bitrates, sizes):
# Utility function (logarithmic)
utility = math.log(bitrate)
# BOLA objective function
objective = (utility + self.Vp * buffer_level) / size
utilities.append(objective)
# Select quality with maximum objective value
max_index = utilities.index(max(utilities))
selected_quality = qualities[max_index]
# Apply safety margins
selected_quality = self._apply_safety_margins(
selected_quality, session_state, available_qualities
)
return selected_quality
def _apply_safety_margins(self, selected_quality: str,
session_state: dict,
available_qualities: List[dict]) -> str:
"""Apply safety margins to prevent oscillations"""
current_quality = session_state.get('current_quality')
if not current_quality:
return selected_quality
# Find indices
quality_names = [q['name'] for q in available_qualities]
current_idx = quality_names.index(current_quality)
selected_idx = quality_names.index(selected_quality)
# Buffer thresholds
buffer_level = session_state.get('buffer_level', 0)
# Conservative upgrades
if selected_idx > current_idx:
# Only upgrade if buffer is healthy
if buffer_level < 10: # 10 seconds
return current_quality
# Limit upgrade steps
max_upgrade = 1 # Only upgrade one step at a time
if selected_idx - current_idx > max_upgrade:
selected_idx = current_idx + max_upgrade
# Aggressive downgrades
elif selected_idx < current_idx:
# Downgrade quickly if buffer is low
if buffer_level < 5: # 5 seconds
# Downgrade multiple steps if needed
while selected_idx > 0 and buffer_level < 5:
selected_idx -= 1
# Recalculate buffer with lower quality
buffer_level += 2 # Simulated improvement
return quality_names[selected_idx]
class ThroughputBasedABR:
"""Throughput-based ABR algorithm"""
def __init__(self, config: dict = None):
self.config = config or {}
self.throughput_window = self.config.get('throughput_window', 5)
self.throughput_history = deque(maxlen=self.throughput_window)
async def select_quality(self, session_state: dict,
available_qualities: List[dict]) -> str:
"""Select quality based on throughput estimation"""
# Estimate throughput
throughput = self._estimate_throughput(session_state)
# Add safety margin
safe_throughput = throughput * 0.8
# Select highest quality that fits in safe throughput
selected_quality = available_qualities[0]['name']
for quality in available_qualities:
if quality['bitrate'] <= safe_throughput:
selected_quality = quality['name']
else:
break
# Consider buffer level
buffer_level = session_state.get('buffer_level', 0)
if buffer_level < 5: # 5 seconds
# Downgrade for safety
quality_names = [q['name'] for q in available_qualities]
current_idx = quality_names.index(selected_quality)
if current_idx > 0:
selected_quality = quality_names[current_idx - 1]
elif buffer_level > 20: # 20 seconds
# Consider upgrade
quality_names = [q['name'] for q in available_qualities]
current_idx = quality_names.index(selected_quality)
if current_idx < len(quality_names) - 1:
next_quality = available_qualities[current_idx + 1]
if next_quality['bitrate'] <= safe_throughput:
selected_quality = next_quality['name']
return selected_quality
def _estimate_throughput(self, session_state: dict) -> float:
"""Estimate network throughput"""
# Get recent download measurements
recent_downloads = session_state.get('recent_downloads', [])
if not recent_downloads:
# Default to lowest quality bitrate
return 100000 # 100 kbps
# Calculate weighted average
weights = []
values = []
for i, download in enumerate(recent_downloads[-self.throughput_window:]):
# More recent measurements get higher weight
weight = 1.0 / (i + 1)
weights.append(weight)
values.append(download['throughput'])
# Weighted average
weighted_sum = sum(w * v for w, v in zip(weights, values))
total_weight = sum(weights)
return weighted_sum / total_weight
class HybridABR:
"""Hybrid ABR combining multiple algorithms"""
def __init__(self):
self.bola = BOLAAlgorithm()
self.throughput = ThroughputBasedABR()
self.mab = MultiArmedBandit() # Machine learning approach
async def select_quality(self, session_state: dict,
available_qualities: List[dict]) -> str:
"""Select quality using hybrid approach"""
# Get suggestions from all algorithms
bola_quality = await self.bola.select_quality(
session_state, available_qualities
)
throughput_quality = await self.throughput.select_quality(
session_state, available_qualities
)
mab_quality = await self.mab.select_quality(
session_state, available_qualities
)
# Weighted combination based on conditions
buffer_level = session_state.get('buffer_level', 0)
network_stability = self._calculate_network_stability(session_state)
if buffer_level < 5:
# Low buffer: prefer conservative approach
weights = {'bola': 0.6, 'throughput': 0.4, 'mab': 0.0}
elif network_stability < 0.7:
# Unstable network: prefer throughput-based
weights = {'bola': 0.3, 'throughput': 0.6, 'mab': 0.1}
else:
# Stable conditions: use all algorithms
weights = {'bola': 0.4, 'throughput': 0.3, 'mab': 0.3}
# Weighted voting
votes = {}
for quality, weight in [(bola_quality, weights['bola']),
(throughput_quality, weights['throughput']),
(mab_quality, weights['mab'])]:
votes[quality] = votes.get(quality, 0) + weight
# Select quality with highest weight
selected_quality = max(votes.items(), key=lambda x: x[1])[0]
return selected_quality
Live Streaming Infrastructure
1. Real-time Media Server
python
# streaming/media_server.py
class MediaServer:
"""Real-time media server for live streaming"""
def __init__(self):
self.ingest_nodes = KubernetesDeployment()
self.edge_nodes = CDNIntegration()
self.transcoding = TranscodingService()
self.webrtc = WebRTCGateway()
async def ingest_live_stream(self, stream_key: str, protocol: str,
input_url: str) -> dict:
"""Ingest live stream from broadcaster"""
# Validate stream key
stream_info = await self._validate_stream_key(stream_key)
# Select ingest node based on broadcaster location
ingest_node = await self._select_ingest_node(
stream_info['broadcaster_location']
)
# Setup ingestion based on protocol
if protocol == 'rtmp':
ingestion = await self._setup_rtmp_ingestion(
ingest_node, stream_info['stream_id'], input_url
)
elif protocol == 'srt':
ingestion = await self._setup_srt_ingestion(
ingest_node, stream_info['stream_id'], input_url
)
elif protocol == 'webrtc':
ingestion = await self._setup_webrtc_ingestion(
ingest_node, stream_info['stream_id'], input_url
)
else:
raise UnsupportedProtocol()
# Start health monitoring
asyncio.create_task(
self._monitor_ingestion_health(
stream_info['stream_id'], ingestion
)
)
return {
'stream_id': stream_info['stream_id'],
'ingestion_url': ingestion['url'],
'playback_urls': await self._generate_playback_urls(
stream_info['stream_id']
),
'latency': ingestion['latency']
}
async def _setup_rtmp_ingestion(self, node: dict, stream_id: str,
input_url: str) -> dict:
"""Setup RTMP ingestion"""
# Configure Nginx RTMP module or similar
config = {
'application': 'live',
'name': stream_id,
'allow': 'publish',
'live': 'on',
'meta': 'copy',
'hls': 'on',
'hls_path': f'/tmp/hls/{stream_id}',
'hls_fragment': '2s',
'hls_playlist_length': '10s',
'dash': 'on',
'dash_path': f'/tmp/dash/{stream_id}',
'dash_fragment': '2s',
'dash_playlist_length': '10s'
}
# Apply configuration
await node.configure_rtmp(config)
return {
'url': f"rtmp://{node['host']}/live/{stream_id}",
'latency': '5-10s',
'protocol': 'rtmp'
}
async def _setup_webrtc_ingestion(self, node: dict, stream_id: str,
input_url: str) -> dict:
"""Setup WebRTC ingestion (ultra-low latency)"""
# Create WebRTC peer connection
peer_connection = await self.webrtc.create_peer_connection()
# Setup media tracks
await peer_connection.add_transceiver('video', {'direction': 'sendonly'})
await peer_connection.add_transceiver('audio', {'direction': 'sendonly'})
# Generate SDP offer
offer = await peer_connection.create_offer()
await peer_connection.set_local_description(offer)
return {
'url': f"wss://{node['host']}/ws/{stream_id}",
'sdp_offer': offer.sdp,
'latency': '500ms',
'protocol': 'webrtc'
}
async def distribute_to_edge(self, stream_id: str, source_url: str):
"""Distribute live stream to edge nodes"""
# Get list of edge nodes
edge_nodes = await self.edge_nodes.get_nodes()
# Select edge nodes based on viewer distribution
selected_nodes = self._select_edge_nodes(
edge_nodes, await self._get_viewer_distribution(stream_id)
)
# Setup edge distribution
distribution_tasks = []
for node in selected_nodes:
task = asyncio.create_task(
self._setup_edge_distribution(node, stream_id, source_url)
)
distribution_tasks.append(task)
# Wait for distribution setup
await asyncio.gather(*distribution_tasks)
# Setup load balancing
await self._setup_load_balancing(stream_id, selected_nodes)
async def _setup_edge_distribution(self, node: dict, stream_id: str,
source_url: str):
"""Setup distribution to edge node"""
# Pull from source
await node.pull_stream(
source_url=source_url,
stream_id=stream_id,
protocols=['hls', 'dash'],
qualities=['source', '1080p', '720p', '480p']
)
# Configure caching
await node.configure_cache(
stream_id=stream_id,
cache_size='1GB',
cache_ttl='30s'
)
# Enable low-latency mode if supported
if node.get('supports_ll_hls'):
await node.configure_ll_hls(
stream_id=stream_id,
partial_segment_duration='0.25s',
hold_back='3.0s',
part_hold_back='1.0s'
)
async def handle_viewer_scaling(self, stream_id: str):
"""Handle scaling based on viewer count"""
while True:
try:
# Get current viewer count
viewer_count = await self._get_viewer_count(stream_id)
# Get current resource allocation
current_allocation = await self._get_resource_allocation(stream_id)
# Calculate desired allocation
desired_allocation = self._calculate_desired_allocation(
viewer_count, current_allocation
)
# Scale if needed
if desired_allocation != current_allocation:
await self._scale_resources(
stream_id, desired_allocation
)
# Check for sudden spikes
if viewer_count > current_allocation['max_viewers'] * 1.5:
# Emergency scale up
await self._emergency_scale_up(stream_id, viewer_count)
await asyncio.sleep(30) # Check every 30 seconds
except Exception as e:
logger.error(f"Error in viewer scaling: {e}")
await asyncio.sleep(60)
2. WebRTC Mesh Network for Low Latency
python
# streaming/webrtc_mesh.py
class WebRTCMesh:
"""WebRTC mesh network for ultra-low latency streaming"""
def __init__(self):
self.signaling_server = SignalingServer()
self.stun_servers = STUNCluster()
self.turn_servers = TURNCluster()
self.mesh_nodes = KubernetesDeployment()
async def create_mesh(self, stream_id: str, broadcaster_peer: dict):
"""Create WebRTC mesh network for a stream"""
# Create mesh topology
topology = await self._create_mesh_topology(stream_id)
# Setup signaling channels
signaling_channels = await self._setup_signaling_channels(
stream_id, topology
)
# Connect broadcaster to initial nodes
initial_connections = await self._connect_broadcaster(
broadcaster_peer, topology['initial_nodes']
)
# Setup ICE servers
ice_servers = await self._get_ice_servers()
# Monitor and optimize mesh
asyncio.create_task(
self._optimize_mesh_topology(stream_id, topology)
)
return {
'mesh_id': topology['id'],
'signaling_channels': signaling_channels,
'ice_servers': ice_servers,
'initial_connections': initial_connections,
'topology': topology
}
async def _create_mesh_topology(self, stream_id: str) -> dict:
"""Create optimized mesh topology"""
# Estimate expected viewer count
expected_viewers = await self._estimate_viewer_count(stream_id)
# Calculate optimal node count
node_count = self._calculate_optimal_node_count(expected_viewers)
# Create mesh nodes
nodes = []
for i in range(node_count):
node = await self._create_mesh_node(f"{stream_id}-node-{i}")
nodes.append(node)
# Create connections between nodes
connections = self._create_mesh_connections(nodes)
# Select edge nodes for viewer connections
edge_nodes = self._select_edge_nodes(nodes)
return {
'id': f"mesh-{stream_id}",
'nodes': nodes,
'connections': connections,
'edge_nodes': edge_nodes,
'expected_viewers': expected_viewers
}
async def _optimize_mesh_topology(self, stream_id: str, topology: dict):
"""Continuously optimize mesh topology"""
while True:
try:
# Get current metrics
metrics = await self._get_mesh_metrics(topology['id'])
# Check for bottlenecks
bottlenecks = self._detect_bottlenecks(metrics)
if bottlenecks:
# Optimize topology
await self._optimize_topology(topology, bottlenecks)
# Check for scaling needs
viewer_count = await self._get_viewer_count(stream_id)
if viewer_count > topology['expected_viewers'] * 1.5:
# Scale up mesh
await self._scale_mesh(topology, viewer_count)
elif viewer_count < topology['expected_viewers'] * 0.5:
# Scale down mesh
await self._downscale_mesh(topology, viewer_count)
await asyncio.sleep(10) # Optimize every 10 seconds
except Exception as e:
logger.error(f"Error optimizing mesh: {e}")
await asyncio.sleep(30)
async def handle_viewer_connection(self, viewer_id: str, stream_id: str):
"""Handle new viewer connection to mesh"""
# Get mesh topology
topology = await self._get_mesh_topology(stream_id)
# Select optimal edge node for viewer
edge_node = await self._select_optimal_edge_node(
viewer_id, topology['edge_nodes']
)
# Create WebRTC connection
connection = await self._create_webrtc_connection(
viewer_id, edge_node
)
# Setup data channel for control messages
data_channel = await connection.create_data_channel('control')
# Setup adaptive bitrate for viewer
await self._setup_viewer_abr(viewer_id, connection)
# Monitor connection quality
asyncio.create_task(
self._monitor_viewer_connection(viewer_id, connection)
)
return {
'connection_id': connection.id,
'edge_node': edge_node['id'],
'sdp_offer': connection.local_description.sdp,
'data_channel': data_channel.label
}
Content Moderation & Copyright System
1. AI-powered Content Moderation
python
# moderation/content_moderation.py
class ContentModeration:
"""AI-powered content moderation system"""
def __init__(self):
self.video_analysis = VideoAnalysisML()
self.audio_analysis = AudioAnalysisML()
self.image_analysis = ImageAnalysisML()
self.text_analysis = TextAnalysisML()
self.human_review = HumanReviewQueue()
async def moderate_video(self, video_id: str, video_path: str) -> dict:
"""Moderate video content using AI"""
moderation_tasks = []
# Video analysis
video_task = asyncio.create_task(
self.video_analysis.analyze(video_path)
)
moderation_tasks.append(video_task)
# Audio analysis
audio_task = asyncio.create_task(
self.audio_analysis.analyze(video_path)
)
moderation_tasks.append(audio_task)
# Extract frames for image analysis
frames = await self._extract_frames(video_path)
image_tasks = []
for frame in frames:
task = asyncio.create_task(
self.image_analysis.analyze(frame)
)
image_tasks.append(task)
# Wait for all analyses
video_result, audio_result = await asyncio.gather(*moderation_tasks)
image_results = await asyncio.gather(*image_tasks)
# Combine results
combined_score = self._combine_moderation_scores(
video_result, audio_result, image_results
)
# Check for violations
violations = self._detect_violations(
video_result, audio_result, image_results
)
# Take action based on severity
if combined_score['severity'] == 'HIGH':
# Immediate takedown
await self._take_down_video(video_id)
action = 'TAKEN_DOWN'
elif combined_score['severity'] == 'MEDIUM':
# Send for human review
await self.human_review.queue_for_review(video_id, combined_score)
action = 'FLAGGED_FOR_REVIEW'
elif combined_score['severity'] == 'LOW':
# Age-restrict if needed
if combined_score.get('age_restriction', False):
await self._age_restrict_video(video_id)
action = 'AGE_RESTRICTED'
else:
action = 'APPROVED'
else:
action = 'APPROVED'
# Store moderation result
await self._store_moderation_result(
video_id, combined_score, violations, action
)
return {
'video_id': video_id,
'moderation_score': combined_score,
'violations': violations,
'action': action,
'review_id': await self._get_review_id(video_id) if action == 'FLAGGED_FOR_REVIEW' else None
}
async def _detect_violations(self, video_result: dict, audio_result: dict,
image_results: List[dict]) -> List[dict]:
"""Detect content policy violations"""
violations = []
# Check for explicit content
if video_result.get('explicit_content', 0) > 0.8:
violations.append({
'type': 'EXPLICIT_CONTENT',
'confidence': video_result['explicit_content'],
'severity': 'HIGH'
})
# Check for violence
if video_result.get('violence', 0) > 0.7:
violations.append({
'type': 'VIOLENCE',
'confidence': video_result['violence'],
'severity': 'HIGH'
})
# Check for hate speech in audio
if audio_result.get('hate_speech', 0) > 0.8:
violations.append({
'type': 'HATE_SPEECH',
'confidence': audio_result['hate_speech'],
'severity': 'HIGH'
})
# Check for copyrighted content
copyright_matches = await self._check_copyright(video_result, audio_result)
if copyright_matches:
violations.append({
'type': 'COPYRIGHT_INFRINGEMENT',
'matches': copyright_matches,
'severity': 'MEDIUM'
})
# Check for spam/misleading content
if video_result.get('spam_score', 0) > 0.9:
violations.append({
'type': 'SPAM',
'confidence': video_result['spam_score'],
'severity': 'MEDIUM'
})
return violations
async def _check_copyright(self, video_result: dict, audio_result: dict) -> List[dict]:
"""Check for copyright infringement"""
matches = []
# Check video fingerprint against database
if video_result.get('fingerprint'):
video_matches = await self._match_video_fingerprint(
video_result['fingerprint']
)
matches.extend(video_matches)
# Check audio fingerprint
if audio_result.get('fingerprint'):
audio_matches = await self._match_audio_fingerprint(
audio_result['fingerprint']
)
matches.extend(audio_matches)
return matches
class CopyrightDetection:
"""Copyright detection using content fingerprinting"""
def __init__(self):
self.fingerprint_db = Cassandra()
self.audio_fingerprint = AcoustID()
self.video_fingerprint = ContentID()
self.rights_database = RightsDB()
async def fingerprint_video(self, video_id: str, video_path: str):
"""Generate and store video fingerprint"""
# Generate video fingerprint
video_fp = await self.video_fingerprint.generate(video_path)
# Generate audio fingerprint
audio_fp = await self.audio_fingerprint.generate(video_path)
# Store fingerprints
await self.fingerprint_db.execute("""
INSERT INTO video_fingerprints
(video_id, video_fingerprint, audio_fingerprint, created_at)
VALUES (?, ?, ?, ?)
""", (video_id, video_fp, audio_fp, datetime.utcnow()))
# Check against known copyrighted content
matches = await self._check_against_copyrighted(video_fp, audio_fp)
if matches:
# Handle copyright matches
await self._handle_copyright_matches(video_id, matches)
return {
'video_id': video_id,
'fingerprint_generated': True,
'matches_found': len(matches) if matches else 0
}
async def _check_against_copyrighted(self, video_fp: str, audio_fp: str):
"""Check fingerprints against copyrighted content database"""
matches = []
# Check video fingerprint
video_matches = await self.fingerprint_db.execute("""
SELECT content_id, confidence, owner_id
FROM copyrighted_fingerprints
WHERE video_fingerprint_similarity(?, video_fingerprint) > 0.8
LIMIT 10
""", (video_fp,))
for match in video_matches:
matches.append({
'type': 'VIDEO',
'content_id': match['content_id'],
'confidence': match['confidence'],
'owner_id': match['owner_id']
})
# Check audio fingerprint
audio_matches = await self.fingerprint_db.execute("""
SELECT content_id, confidence, owner_id
FROM copyrighted_fingerprints
WHERE audio_fingerprint_similarity(?, audio_fingerprint) > 0.8
LIMIT 10
""", (audio_fp,))
for match in audio_matches:
matches.append({
'type': 'AUDIO',
'content_id': match['content_id'],
'confidence': match['confidence'],
'owner_id': match['owner_id']
})
return matches
async def _handle_copyright_matches(self, video_id: str, matches: List[dict]):
"""Handle copyright matches"""
for match in matches:
# Check rights holder's policy
policy = await self.rights_database.get_policy(match['owner_id'])
if policy['action'] == 'BLOCK':
# Block video globally
await self._block_video(video_id, match)
elif policy['action'] == 'MONETIZE':
# Claim ad revenue
await self._claim_revenue(video_id, match)
elif policy['action'] == 'TRACK':
# Just track, no action
await self._track_viewership(video_id, match)
elif policy['action'] == 'BLOCK_IN_REGIONS':
# Block in specific regions
await self._block_in_regions(video_id, match, policy['regions'])
Analytics & Monitoring
1. Real-time Analytics Pipeline
python
# analytics/realtime_analytics.py
class RealTimeAnalytics:
"""Real-time analytics for video streaming"""
def __init__(self):
self.kafka = KafkaConsumer()
self.flink = FlinkCluster()
self.druid = DruidCluster()
self.redis = RedisCluster()
async def process_streaming_events(self):
"""Process real-time streaming events"""
# Consume events from Kafka
consumer = await self.kafka.consume('streaming_events')
async for message in consumer:
event = json.loads(message.value.decode())
# Process based on event type
if event['type'] == 'PLAYBACK_STARTED':
await self._handle_playback_started(event)
elif event['type'] == 'PLAYBACK_PROGRESS':
await self._handle_playback_progress(event)
elif event['type'] == 'PLAYBACK_COMPLETED':
await self._handle_playback_completed(event)
elif event['type'] == 'QUALITY_CHANGED':
await self._handle_quality_changed(event)
elif event['type'] == 'BUFFERING_OCCURRED':
await self._handle_buffering_event(event)
elif event['type'] == 'ERROR_OCCURRED':
await self._handle_error_event(event)
async def _handle_playback_started(self, event: dict):
"""Handle playback started event"""
# Update real-time viewer count
await self._update_viewer_count(event['video_id'], +1)
# Update session start
await self._record_session_start(event)
# Update geo distribution
await self._update_geo_distribution(
event['video_id'], event['location']
)
# Update device distribution
await self._update_device_distribution(
event['video_id'], event['device_type']
)
async def _handle_playback_progress(self, event: dict):
"""Handle playback progress event"""
# Update watch time
await self._update_watch_time(
event['video_id'], event['watch_duration']
)
# Update concurrent viewers
await self._update_concurrent_viewers(
event['video_id'], event['timestamp']
)
# Check for drop-off points
await self._analyze_drop_off(event)
# Update engagement metrics
await self._update_engagement_metrics(event)
async def _handle_playback_completed(self, event: dict):
"""Handle playback completed event"""
# Update viewer count
await self._update_viewer_count(event['video_id'], -1)
# Record completion
await self._record_completion(event)
# Update completion rate
await self._update_completion_rate(event['video_id'])
# Update average watch time
await self._update_average_watch_time(event['video_id'])
async def calculate_real_time_metrics(self, video_id: str) -> dict:
"""Calculate real-time metrics for a video"""
# Get from cache first
cache_key = f"metrics:{video_id}:realtime"
cached = await self.redis.get(cache_key)
if cached:
return json.loads(cached)
# Calculate metrics
metrics = {
'current_viewers': await self._get_current_viewers(video_id),
'total_views_today': await self._get_total_views_today(video_id),
'average_watch_time': await self._get_average_watch_time(video_id),
'completion_rate': await self._get_completion_rate(video_id),
'geo_distribution': await self._get_geo_distribution(video_id),
'device_distribution': await self._get_device_distribution(video_id),
'quality_distribution': await self._get_quality_distribution(video_id),
'buffering_rate': await self._get_buffering_rate(video_id),
'error_rate': await self._get_error_rate(video_id),
'peak_concurrent': await self._get_peak_concurrent(video_id),
'trend': await self._get_viewer_trend(video_id)
}
# Cache for 30 seconds
await self.redis.setex(cache_key, 30, json.dumps(metrics))
return metrics
async def detect_anomalies(self):
"""Detect anomalies in real-time metrics"""
while True:
try:
# Get all active videos
active_videos = await self._get_active_videos()
for video_id in active_videos:
# Get metrics
metrics = await self.calculate_real_time_metrics(video_id)
# Check for anomalies
anomalies = await self._check_for_anomalies(
video_id, metrics
)
if anomalies:
# Trigger alerts
await self._trigger_alerts(video_id, anomalies)
# Take corrective actions if needed
await self._take_corrective_actions(
video_id, anomalies
)
await asyncio.sleep(60) # Check every minute
except Exception as e:
logger.error(f"Error in anomaly detection: {e}")
await asyncio.sleep(30)
2. A/B Testing Framework
python
# analytics/ab_testing.py
class ABTesting:
"""A/B testing framework for video streaming"""
def __init__(self):
self.experiments_db = PostgreSQL()
self.assignment_cache = Redis()
self.analytics = AnalyticsService()
async def assign_experiment(self, user_id: str, context: dict) -> dict:
"""Assign user to experiments"""
# Get active experiments
experiments = await self._get_active_experiments()
assignments = {}
for experiment in experiments:
# Check if user is eligible
if await self._is_eligible(user_id, experiment, context):
# Assign to variant
variant = await self._assign_variant(
user_id, experiment, context
)
assignments[experiment['id']] = {
'experiment_name': experiment['name'],
'variant': variant['name'],
'parameters': variant['parameters']
}
# Cache assignments
await self.assignment_cache.setex(
f"experiments:{user_id}",
86400, # 24 hours
json.dumps(assignments)
)
return assignments
async def track_experiment_event(self, user_id: str, event_type: str,
event_data: dict):
"""Track event for A/B testing"""
# Get user's experiment assignments
assignments = await self.assignment_cache.get(f"experiments:{user_id}")
if not assignments:
assignments = await self.assign_experiment(user_id, {})
else:
assignments = json.loads(assignments)
# Track event for each experiment
for exp_id, assignment in assignments.items():
await self.analytics.track_experiment_event(
experiment_id=exp_id,
variant=assignment['variant'],
user_id=user_id,
event_type=event_type,
event_data=event_data,
timestamp=datetime.utcnow().isoformat()
)
async def analyze_experiment(self, experiment_id: str) -> dict:
"""Analyze experiment results"""
# Get experiment configuration
experiment = await self._get_experiment(experiment_id)
# Get metrics for each variant
variant_metrics = {}
for variant in experiment['variants']:
metrics = await self._calculate_variant_metrics(
experiment_id, variant['name']
)
variant_metrics[variant['name']] = metrics
# Calculate statistical significance
significance = await self._calculate_significance(
variant_metrics, experiment['primary_metric']
)
# Determine winner
winner = await self._determine_winner(
variant_metrics, experiment['primary_metric'], significance
)
return {
'experiment_id': experiment_id,
'status': experiment['status'],
'total_participants': await self._get_total_participants(experiment_id),
'start_date': experiment['start_date'],
'end_date': experiment['end_date'],
'primary_metric': experiment['primary_metric'],
'variant_metrics': variant_metrics,
'statistical_significance': significance,
'winner': winner,
'recommendation': self._generate_recommendation(
winner, significance
)
}
async def run_ab_test(self, video_id: str, test_config: dict):
"""Run A/B test for video features"""
# Create experiment
experiment = await self._create_experiment(
name=f"video_{video_id}_{test_config['name']}",
description=test_config.get('description', ''),
variants=test_config['variants'],
primary_metric=test_config['primary_metric'],
secondary_metrics=test_config.get('secondary_metrics', []),
target_audience=test_config.get('target_audience', {}),
sample_size=test_config.get('sample_size', 10000),
duration_days=test_config.get('duration_days', 7)
)
# Start experiment
await self._start_experiment(experiment['id'])
# Monitor experiment
asyncio.create_task(
self._monitor_experiment(experiment['id'])
)
return experiment
Infrastructure & Cost Optimization
1. Storage Optimization with Tiering
python
# storage/tiered_storage.py
class TieredStorage:
"""Tiered storage system for cost optimization"""
def __init__(self):
self.hot_storage = S3Standard()
self.warm_storage = S3InfrequentAccess()
self.cold_storage = S3Glacier()
self.archive_storage = S3DeepArchive()
self.cache = Redis()
async def optimize_storage(self, video_id: str, access_patterns: dict):
"""Optimize storage based on access patterns"""
# Analyze access patterns
analysis = await self._analyze_access_patterns(video_id, access_patterns)
# Determine optimal storage tier
optimal_tier = self._determine_optimal_tier(analysis)
# Move to optimal tier if needed
current_tier = await self._get_current_tier(video_id)
if current_tier != optimal_tier:
await self._move_to_tier(video_id, optimal_tier)
# Setup lifecycle policies
await self._setup_lifecycle_policies(video_id, analysis)
# Optimize encoding based on tier
await self._optimize_encoding_for_tier(video_id, optimal_tier)
def _determine_optimal_tier(self, analysis: dict) -> str:
"""Determine optimal storage tier based on analysis"""
views_per_day = analysis.get('views_per_day', 0)
last_access_days = analysis.get('last_access_days', 365)
predicted_future_views = analysis.get('predicted_future_views', 0)
if views_per_day > 1000 or predicted_future_views > 500:
return 'HOT' # S3 Standard
elif views_per_day > 100:
return 'WARM' # S3 Infrequent Access
elif last_access_days < 90:
return 'COLD' # S3 Glacier
else:
return 'ARCHIVE' # S3 Deep Archive
async def _move_to_tier(self, video_id: str, target_tier: str):
"""Move video to target storage tier"""
current_tier = await self._get_current_tier(video_id)
if current_tier == target_tier:
return
# Get all files for video
files = await self._list_video_files(video_id)
# Move each file
for file in files:
source_key = file['key']
# Determine target storage class
if target_tier == 'HOT':
storage_class = 'STANDARD'
target_bucket = 'video-hot'
elif target_tier == 'WARM':
storage_class = 'STANDARD_IA'
target_bucket = 'video-warm'
elif target_tier == 'COLD':
storage_class = 'GLACIER'
target_bucket = 'video-cold'
else: # ARCHIVE
storage_class = 'DEEP_ARCHIVE'
target_bucket = 'video-archive'
# Copy to new location
await self._copy_file(
source_bucket=file['bucket'],
source_key=source_key,
target_bucket=target_bucket,
target_key=source_key,
storage_class=storage_class
)
# Delete from old location (after verification)
await self._delete_file(file['bucket'], source_key)
# Update metadata
await self._update_storage_tier(video_id, target_tier)
async def smart_cache_warming(self, video_id: str):
"""Smart cache warming based on predictions"""
# Predict future demand
predictions = await self._predict_demand(video_id)
# Determine what to cache
cache_plan = self._create_cache_plan(video_id, predictions)
# Warm cache
for item in cache_plan['to_cache']:
await self._warm_cache_item(
video_id, item['quality'], item['region']
)
# Pre-transcode if needed
if predictions.get('expected_peak') > 10000:
await self._pre_transcode_popular_qualities(video_id)
# Update CDN configurations
await self._update_cdn_config(video_id, cache_plan)
2. Cost-Aware Transcoding
python
# transcoding/cost_optimizer.py
class CostAwareTranscoder:
"""Cost-aware video transcoding"""
def __init__(self):
self.spot_instances = EC2Spot()
self.reserved_instances = EC2Reserved()
self.on_demand_instances = EC2OnDemand()
self.batch_jobs = AWSBatch()
async def transcode_cost_effectively(self, video_id: str,
video_path: str, qualities: List[dict]):
"""Transcode video using cost-effective resources"""
# Estimate processing cost
cost_estimate = await self._estimate_processing_cost(
video_path, qualities
)
# Check video popularity
popularity = await self._estimate_popularity(video_id)
# Select processing strategy
if popularity == 'HIGH':
# Use reserved instances for guaranteed performance
strategy = 'RESERVED'
instances = await self.reserved_instances.acquire(10)
elif popularity == 'MEDIUM':
# Use spot instances with fallback
strategy = 'SPOT_WITH_FALLBACK'
instances = await self._acquire_spot_with_fallback(5)
else:
# Use batch jobs for low priority
strategy = 'BATCH'
instances = await self.batch_jobs.submit(
self._create_batch_job(video_id, video_path, qualities)
)
# Process with selected strategy
results = await self._process_with_strategy(
video_id, video_path, qualities, strategy, instances
)
# Record cost
actual_cost = await self._calculate_actual_cost(instances, results)
return {
'video_id': video_id,
'strategy': strategy,
'estimated_cost': cost_estimate,
'actual_cost': actual_cost,
'savings': cost_estimate - actual_cost,
'processing_time': results['processing_time']
}
async def _estimate_processing_cost(self, video_path: str,
qualities: List[dict]) -> float:
"""Estimate processing cost"""
# Get video duration
duration = await self._get_video_duration(video_path)
# Calculate processing time per quality
total_processing_seconds = 0
for quality in qualities:
# Estimate based on resolution and codec
processing_time = self._estimate_quality_processing_time(
duration, quality
)
total_processing_seconds += processing_time
# Convert to instance hours
instance_hours = total_processing_seconds / 3600
# Calculate cost
spot_cost = instance_hours * 0.10 # $0.10 per hour
reserved_cost = instance_hours * 0.30 # $0.30 per hour
on_demand_cost = instance_hours * 0.50 # $0.50 per hour
# Return minimum expected cost
return min(spot_cost, reserved_cost, on_demand_cost)
async def optimize_transcoding_pipeline(self):
"""Optimize transcoding pipeline for cost"""
# Analyze queue
queue_analysis = await self._analyze_transcoding_queue()
# Group jobs by priority and requirements
job_groups = self._group_jobs_by_requirements(queue_analysis['jobs'])
# Allocate resources optimally
for group in job_groups:
if group['priority'] == 'HIGH':
# Use reserved instances
await self._process_with_reserved(group['jobs'])
elif group['priority'] == 'MEDIUM':
# Use spot instances
await self._process_with_spot(group['jobs'])
else:
# Use batch during off-peak
await self._schedule_for_off_peak(group['jobs'])
# Monitor and adjust
await self._monitor_and_adjust_allocation()
Key Takeaways & Best Practices
1. Architecture Principles:
-
Microservices: Separate upload, processing, streaming, recommendation
-
Event-Driven: Kafka for async communication between services
-
CQRS: Separate read/write models for scalability
-
CDN First: Edge delivery for global low latency
-
Multi-Format: Support HLS, DASH, WebRTC for different use cases
2. Performance Optimizations:
-
Adaptive Bitrate: Multiple algorithms (BOLA, throughput-based, hybrid)
-
Smart Caching: Tiered caching at edge, region, origin
-
Connection Pooling: Reuse WebRTC/SRT connections
-
Protocol Optimization: Choose based on latency requirements
-
Pre-fetching: Predict and pre-load content
3. Cost Optimization:
-
Storage Tiering: Hot, warm, cold, archive based on access patterns
-
Compute Spot Instances: For non-critical transcoding
-
CDN Optimization: Regional caching, origin shielding
-
Transcoding Efficiency: Reuse intermediate files, smart quality selection
-
Data Transfer: Minimize cross-region data transfer
4. Reliability & Scalability:
-
Multi-CDN: Fallback across CDN providers
-
Auto-scaling: Based on viewer count and geographic distribution
-
Circuit Breakers: Prevent cascading failures
-
Graceful Degradation: Lower quality during issues
-
Disaster Recovery: Cross-region replication
5. Monetization Strategies:
-
Ad Insertion: Server-side and client-side ad insertion
-
Subscription Tiers: Multiple quality levels, exclusive content
-
Pay-per-View: For premium live events
-
Sponsorships: Brand integrations, product placements
-
Merchandising: Integrated shopping during streams
6. Compliance & Security:
-
DRM: Widevine, FairPlay, PlayReady for content protection
-
GDPR Compliance: User data protection, right to be forgotten
-
Content Moderation: AI + human review for policy compliance
-
Copyright Management: Fingerprinting, automated claims
-
Access Control: Region restrictions, age gating
7. Analytics & Insights:
-
Real-time Metrics: Viewer count, engagement, drop-off points
-
A/B Testing: Experiment with features, algorithms, UI
-
Predictive Analytics: Demand forecasting, recommendation optimization
-
Quality of Experience: Buffer ratio, startup time, rebuffering events
-
Business Intelligence: Revenue, retention, growth metrics
This system can handle 1B+ daily users, 1M+ concurrent streams, petabytes of daily video uploads, and millions of real-time interactions while maintaining sub-second latency for live streaming and adaptive quality for varying network conditions.
Comments (0)
No comments yet. Be the first to comment!