Ads Audience Targeting / Custom Audience System

Design a system where advertisers upload first-party user lists (emails / phones / device IDs), match them to internal users, package them with AND/OR/NOT rules, and answer membership checks in under 10ms at 200M-user scale. Design an Ads Audience Targeting System Design a system that lets advertisers upload lists of users. The system must group these users into segments and use them to target ads precisely. The system handles "first-party data"—this means data the advertiser owns, like email lists or customer IDs. The main challenge is speed and size. You need to process files with millions of users quickly. Then, when an ad request comes in, you must check if a user is in a group in under 10ms. This system needs to work at Netflix's scale (200M+ users) and stay online all the time. Phase 1: Project Requirements What the System Must Do Advertisers upload files: Support large files (CSV, JSON) with millions of user IDs. System matches users: Match the uploaded data (emails, phone numbers) to our internal user base. Create packages: Allow advertisers to combine groups using rules like AND/OR/NOT. Fast checks: The ad server must quickly check if a user belongs to a group. Track progress: Advertisers can see upload progress, match rates, and errors. Note: Features like "lookalike audiences" or real-time updates are "nice-to-have" but not the main focus here. Performance Goals Requirement Target Why? Large file support 10M+ IDs, up to 1GB Customer lists are huge. Processing time 10M rows in < 30 mins Advertisers need to launch campaigns fast. Lookup latency Sub-10ms Ad servers must respond instantly. Availability 99.9% uptime Ad serving cannot stop. Fault tolerance Resume failed uploads Big uploads often fail halfway; don't lose progress. Data privacy Hashed IDs only We must protect user privacy and follow laws. Important: For uploads, throughput (speed of processing) matters most. Advertisers can wait 30 minutes. But for ad serving, latency (speed of response) matters most. The system cannot take 100ms to check a user. Capacity Estimates Metric Value Total users 200M Active advertisers 10K Total audience groups 100K Average group size 200K users Ad requests 100K/second Membership lookups 500K-1M/second Storage Math: One membership link: user_id (8B) + audience_id (8B) = 16 bytes. Total links: 100K groups × 200K users = 20B links. Raw size: 20B × 16 bytes = 320 GB. With database overhead (Cassandra) and copies for safety (Replication Factor=3), we need multiple TB of storage. We will use Cassandra for storage and Redis for caching. Phase 2: Data Model Database Schema Audience ├── id: UUID (PK) ├── advertiser_id: UUID (FK) ├── name: VARCHAR ├── type: ENUM (uploaded, rule_based, package) ├── status: ENUM (processing, ready, failed, expired) ├── user_count: INTEGER ├── match_rate: DECIMAL ├── created_at: TIMESTAMP ├── updated_at: TIMESTAMP └── expires_at: TIMESTAMP AudienceUpload ├── id: UUID (PK) ├── audience_id: UUID (FK) ├── file_path: VARCHAR ├── file_size_bytes: BIGINT ├── total_rows: INTEGER ├── processed_rows: INTEGER ├── matched_rows: INTEGER ├── status: ENUM (pending, processing, completed, failed) ├── error_message: VARCHAR ├── checkpoint_offset: BIGINT ├── started_at: TIMESTAMP └── completed_at: TIMESTAMP AudiencePackage ├── id: UUID (PK) ├── advertiser_id: UUID (FK) ├── name: VARCHAR ├── expression: JSONB -- {"operator": "AND", "operands": [...]} ├── estimated_size: INTEGER └── created_at: TIMESTAMP AudienceMembership (Hot path - Cassandra, dual tables for different access patterns) -- Primary table: Fast lookups by user during ad serving memberships_by_user ├── user_id: VARCHAR (partition key) ├── audience_id: UUID (clustering key) └── added_at: TIMESTAMP -- Reverse index: Efficient deletion and stats by audience memberships_by_audience ├── audience_id: UUID (partition key) ├── user_id: VARCHAR (clustering key) └── added_at: TIMESTAMP IdentifierMapping (Pre-built index; one identifier can map to multiple users) ├── identifier_type: ENUM (email_sha256, phone_sha256, device_id) ├── identifier_value: VARCHAR (hashed) └── user_id: VARCHAR -- multiple rows per identifier_value Matching User Data Advertisers upload different types of data: Emails (Hashed with SHA256) Phone numbers (Hashed with SHA256) Device IDs We keep a lookup table that maps these external IDs to our internal user IDs. This table is updated daily. Privacy Rule: Never store raw personal data. Advertisers must hash data before sending, or we hash it immediately and delete the raw version. We only store hashes. This lets us match users without seeing their actual emails. Phase 3: API Design Managing Audiences # Create audience POST /api/v1/audiences Request: { "name": "Holiday Shoppers 2024", "description": "Users who purchased during holiday season" } Response: { "audience_id": "aud_abc123", "status": "created" } # Get audience details GET /api/v1/audiences/{audience_id} Response: { "audience_id": "aud_abc123", "name": "Holiday Shoppers 2024", "status": "ready", "user_count": 2500000, "match_rate": 0.72, "created_at": "2024-01-15T10:00:00Z" } Uploading Files (Resumable) # Initiate multipart upload POST /api/v1/audiences/{audience_id}/uploads/initiate Request: { "file_name": "customers.csv", "file_size": 524288000, "identifier_type": "email_sha256" } Response: { "upload_id": "upl_xyz789", "part_size": 10485760, "total_parts": 50, "upload_urls": [ {"part_number": 1, "url": "https://s3.../part1?presigned..."}, {"part_number": 2, "url": "https://s3.../part2?presigned..."} ] } # Complete upload and trigger processing POST /api/v1/audiences/{audience_id}/uploads/{upload_id}/complete Request: { "parts": [ {"part_number": 1, "etag": "abc123"}, {"part_number": 2, "etag": "def456"} ] } Response: { "upload_id": "upl_xyz789", "status": "processing", "estimated_completion": "2024-01-15T10:30:00Z" } # Check upload/processing status GET /api/v1/audiences/{audience_id}/uploads/{upload_id}/status Response: { "upload_id": "upl_xyz789", "status": "processing", "progress": { "total_rows": 10000000, "processed_rows": 7500000, "matched_rows": 5400000, "percentage": 75 } } Why use presigned URLs? This lets the client upload directly to S3 storage. It saves our API servers from handling huge amounts of data traffic. Audience Package API POST /api/v1/audience-packages Request: { "name": "High-Value Non-Subscribers", "expression": { "operator": "AND", "operands": [ {"audience_id": "aud_high_value"}, { "operator": "NOT", "operands": [{"audience_id": "aud_subscribers"}] } ] } } Response: { "package_id": "pkg_123", "estimated_size": 1500000, "status": "ready" } Checking User Membership (Internal) // gRPC for low latency service AudienceService { rpc CheckMembership(MembershipRequest) returns (MembershipResponse); } message MembershipRequest { string user_id = 1; repeated string audience_ids = 2; } message MembershipResponse { map<string, bool> memberships = 1; // audience_id -> is_member } Phase 4: High-Level Design System Architecture What Each Part Does: Audience API: Handles requests and tracks status. S3 / Blob Storage: Holds the uploaded files. Processing Queue: Holds jobs so we can retry them if they fail. Processing Workers: Programs that read files, match users, and save data. Match Service: Finds the internal user ID for an email hash. Identifier Index: A map of Hash -> User ID. Membership Store (Cassandra): The main database for user groups. Redis Cluster: A fast cache for ad serving. Lookup Service: The service ad servers call to check users. How Uploads Work Chunk-Based Processing: We break large files into small chunks. This helps us: Track Progress: We can say "50% done". Resume: If it crashes, we restart from the last saved spot. Save Memory: We don't load the whole 1GB file at once. Speed: We can process many chunks at the same time. def process_upload(upload_id: str, audience_id: str, file_path: str): chunk_size = 10_000 checkpoint = get_checkpoint(upload_id) # Resume from failure for chunk in read_file_chunks(file_path, chunk_size, start=checkpoint): # 1. Match identifiers to user IDs identifiers = extract_identifiers(chunk) user_ids = flatten(match_service.batch_match(identifiers)) # May return multiple users per identifier # 2. Write memberships memberships = [(uid, audience_id) for uid in user_ids if uid] membership_db.batch_insert(memberships) # 3. Update checkpoint save_checkpoint(upload_id, chunk.end_offset) update_progress(upload_id, chunk.row_count) # 4. Finalize update_audience_stats(audience_id) warm_cache(audience_id) Checkpointing: We save our spot every 10K-100K rows. Saving too often slows us down; saving too rarely makes retries painful. How Ad Checks Work Caching: We use a "cache-aside" strategy with a 1-hour TTL (Time To Live). If a user joins a group, it might take 1 hour to update in ads. This is okay for advertising. Updating Audiences Advertisers often need to add or remove users from an existing group: # Add users POST /api/v1/audiences/{audience_id}/members Request: { "action": "add", "identifiers": [...] } # Remove users POST /api/v1/audiences/{audience_id}/members Request: { "action": "remove", "identifiers": [...] } Logic: Match the ID to a user. Add: Insert into Cassandra. Remove: Delete from Cassandra. Cache: Let the cache expire naturally (1 hour) or force delete it. For removing users (e.g., "Don't show ads to these people"), it is safer to force-clear the cache immediately. Cleaning Up Old Audiences Audiences expire after 90-180 days. A background job cleans them up: def cleanup_expired_audiences(): # 1. Find expired audiences expired = db.query("SELECT id FROM audiences WHERE expires_at < NOW()") for audience_id in expired: # 2. Get user list from reverse index, then delete memberships user_ids = session.execute( "SELECT user_id FROM memberships_by_audience WHERE audience_id = ?", [audience_id] ) # Batch delete from primary table (one delete per partition) delete_stmt = session.prepare( "DELETE FROM memberships_by_user WHERE user_id = ? AND audience_id = ?" ) for user_batch in chunk(user_ids, 100): batch = BatchStatement() for user_id in user_batch: batch.add(delete_stmt, (user_id, audience_id)) session.execute(batch) # 3. Delete from reverse index session.execute( "DELETE FROM memberships_by_audience WHERE audience_id = ?", [audience_id] ) # 4. Mark audience as expired db.execute("UPDATE audiences SET status = 'expired' WHERE id = ?", [audience_id]) # 5. Delete uploaded files from S3 s3.delete_objects(prefix=f"uploads/{audience_id}/") Data Model Note: We use two tables. One is organized by user_id (fast for lookups). The other is organized by audience_id (fast for deleting the whole group). Phase 5: Scaling & Trade-offs Handling High Traffic We need to handle 1 million lookups per second. We use layers of caching. Tier Type Speed TTL Purpose L1 Local Memory Microseconds 5 min Super hot data on the server itself. L2 Redis Cluster ~1ms 1 hour Shared cache for everything. L3 Cassandra ~10ms Permanent The main database if cache misses. def check_membership(user_id: str, audience_ids: list[str]) -> dict: results = {} remaining = audience_ids.copy() # L1: Check local cache for aud_id in remaining[:]: if (user_id, aud_id) in local_cache: results[aud_id] = local_cache[(user_id, aud_id)] remaining.remove(aud_id) if not remaining: return results # L2: Check Redis (batch) redis_keys = [f"mem:{user_id}:{aud_id}" for aud_id in remaining] redis_results = redis.mget(redis_keys) for aud_id, result in zip(remaining[:], redis_results): if result is not None: results[aud_id] = result == "1" remaining.remove(aud_id) if not remaining: return results # L3: Query Cassandra, backfill caches db_results = cassandra.query_memberships(user_id, remaining) for aud_id in remaining: results[aud_id] = aud_id in db_results redis.setex(f"mem:{user_id}:{aud_id}", 3600, "1" if results[aud_id] else "0") return results Matching Millions of Users Challenge: Matching 10M uploaded IDs against 200M users is slow. Solution: Pre-build an index. We build a map: Hash -> [User IDs]. We store this in a fast lookup service. Why? Checking this index is instant (O(1)). Joining tables in real-time takes hours. Edge Case: If a new user signs up today, they might not be in the index until tomorrow. This is usually acceptable for ads. Handling Failures If Uploads Fail: Checkpointing: We save progress so we don't start over. Retry: Workers automatically retry if there is a glitch. If Lookups Fail: Circuit Breaker: If the database is overwhelmed, stop asking it. Fail-Open (Targeting): Assume the user is NOT in the group. The ad still runs, just without targeting. Fail-Closed (Suppression): If the rule is "Do NOT show ads to these people," assume the user IS in the group. This prevents showing ads by mistake. def check_membership_with_fallback(user_id: str, audience_ids: list[str], suppression_ids: set[str]) -> dict: # Skip lookup entirely if circuit is open (too many recent failures) if circuit_breaker.is_open(): metrics.increment("audience_lookup_circuit_open") return { aud_id: (aud_id in suppression_ids) for aud_id in audience_ids } try: result = lookup_service.check_membership(user_id, audience_ids) circuit_breaker.record_success() return result except LookupServiceError: metrics.increment("audience_lookup_failure") circuit_breaker.record_failure() # Targeting: fail open (assume not in audience). Suppression: fail closed. return { aud_id: (aud_id in suppression_ids) for aud_id in audience_ids } Evaluating Packages (AND/OR Logic) We can pre-calculate complex rules or check them in real-time. Recommendation: A hybrid approach. Simple rules (A AND B) are checked in real-time. We look up A and B in parallel, then combine the results. def evaluate_package(user_id: str, expression: dict) -> bool: # Collect all audience IDs from the expression tree audience_ids = extract_audience_ids(expression) # Batch lookup for all audiences (single call, not N sequential calls) memberships = check_membership(user_id, audience_ids) # Evaluate boolean expression with results return evaluate_expression(expression, memberships) Fixing Slow Points Slow Uploads: Use parallel workers and batch writes to the database. Memory Usage: Shard Redis (split data across many servers). Huge Index: Use Bloom filters to quickly skip non-matches. Review Checklist Requirements Did you allow large file uploads? Did you focus on high speed for lookups (sub-10ms)? Did you plan for storage scale (TB of data)? Data Model Did you design the Database tables correctly? Did you explain why we need two Cassandra tables? Did you handle privacy (hashing)? API Design Did you use presigned URLs for uploading? Is there a way to track progress? Did you use gRPC for internal speed? High-Level Design Is the upload flow clear (S3 -> Queue -> Worker)? Is the lookup flow clear (Ad Server -> Cache -> DB)? Did you explain chunking and checkpoints? Did you cover deleting old data? Scaling Did you explain the caching layers (L1/L2/L3)? Did you explain the pre-built index? Did you discuss "Fail-open" vs "Fail-closed"? Important Takeaways Speed matters for uploads: Advertisers shouldn't wait hours. Latency matters for ads: You must answer in under 10ms. Plan for failure: Big files fail. Use checkpoints to resume them. Use an index: Don't join massive tables in real-time. Pre-build the map. Safety first: Fail safely. If a "block list" check fails, block the ad just in case. Privacy: Only store hashes, never raw personal info.