滑动窗口唯一计数方法咨询：IP-用户分钟级登录记录分析需求

Alright, let's tackle this sliding window unique counting problem for your IP-user login tracking scenario. I'll walk you through practical implementation approaches tailored to your minute-granularity data:

Before diving into code, let's align on what we're building: For a defined sliding time window (e.g., 5 minutes), we need to count unique users per IP address that logged in within that window. The window slides forward at a set interval (e.g., every 1 minute), so we'll have a fresh count for each interval.

2.1 Data Preprocessing: Structurize Raw Logs #

Your raw data is unstructured, so first we need to parse each entry into a structured format with three fields:

• minute: Numeric representation of time (e.g., 0 for 00:00, 1 for 00:01, up to 1439 for 23:59)
• ip: The user's IP address
• username: The unique username (stripped of quotes)

Here's a quick Python snippet to parse a single raw line:

raw_entry = '00:00 1.1.1.1 "Ben"'
time_str, ip, username = raw_entry.split(' ', 2)
username = username.strip('"')
# Convert time to total minutes since midnight
total_minutes = int(time_str.split(':')[0]) * 60 + int(time_str.split(':')[1])

Pro tip: Preprocess all entries first, and deduplicate any (minute, ip, username) triples—since multiple logins from the same user on the same IP in the same minute shouldn't count multiple times.

2.2 Real-Time Processing: Hash Table + Timestamp Tracking #

If you need to process logs as they come (real-time), this approach is lightweight and efficient:

How it works: #

• Maintain a dictionary for each IP, tracking two things:
  • A sub-dictionary mapping usernames to their last seen minute (within the window)
  • A running count of unique users in the current window
• For each new log entry:
  1. Clean up expired users: Remove any users from the IP's tracking dict whose last seen minute is outside the current sliding window.
  2. Update counts: If the user isn't in the current window's tracked users, add them and increment the count; if they're already present, do nothing.

Example Code: #

from collections import defaultdict

# Configure window parameters
WINDOW_SIZE = 5  # 5-minute window
SLIDE_INTERVAL = 1  # Slide every 1 minute

# Initialize tracker: key = IP, value = {"users": {username: last_seen_minute}, "count": int}
ip_tracker = defaultdict(lambda: {"users": {}, "count": 0})

def clean_expired_users(current_minute):
    """Remove users who've fallen outside the sliding window for all IPs"""
    window_start = current_minute - WINDOW_SIZE + 1
    for ip in list(ip_tracker.keys()):  # Iterate over copy to avoid modification issues
        tracker = ip_tracker[ip]
        expired_users = [user for user, ts in tracker["users"].items() if ts < window_start]
        for user in expired_users:
            del tracker["users"][user]
            tracker["count"] -= 1
        # Remove empty IP trackers to save memory
        if tracker["count"] == 0:
            del ip_tracker[ip]

def process_log_entry(minute, ip, username):
    clean_expired_users(minute)
    tracker = ip_tracker[ip]
    window_start = minute - WINDOW_SIZE + 1

    # Check if user is already in the current window
    if username not in tracker["users"] or tracker["users"][username] < window_start:
        tracker["users"][username] = minute
        tracker["count"] += 1

    # Output the current unique count for this IP
    print(f"[{minute:04d}] IP {ip}: {tracker['count']} unique users in {WINDOW_SIZE}-min window")

2.3 Offline Batch Processing: Sort + Two Pointers #

If you're processing a full day's logs after the fact, this method is more efficient for large datasets:

How it works: #

1. Group and sort: Group all log entries by IP, then sort each group by minute.
2. Two-pointer sliding window: For each IP's sorted list, use left/right pointers to maintain the window bounds. Use a set to track unique users in the window, and a counter to track how many times each user appears (to avoid prematurely removing users who have multiple entries in the window).

Example Code: #

from collections import defaultdict

def batch_sliding_window_count(processed_records, window_size):
    # Group records by IP, then sort each group by minute
    ip_groups = defaultdict(list)
    for minute, ip, username in processed_records:
        ip_groups[ip].append((minute, username))
    
    # Sort each IP's records by minute
    for ip in ip_groups:
        ip_groups[ip].sort(key=lambda x: x[0])
    
    results = {}
    for ip in ip_groups:
        sorted_records = ip_groups[ip]
        n = len(sorted_records)
        left = 0
        user_set = set()
        user_occurrences = defaultdict(int)
        ip_results = []

        for right in range(n):
            current_minute, user = sorted_records[right]
            # Add current user to tracking
            user_set.add(user)
            user_occurrences[user] += 1

            # Move left pointer to maintain window bounds
            window_start = current_minute - window_size + 1
            while sorted_records[left][0] < window_start:
                left_minute, left_user = sorted_records[left]
                user_occurrences[left_user] -= 1
                # Only remove from set if no more occurrences in window
                if user_occurrences[left_user] == 0:
                    user_set.remove(left_user)
                left += 1

            # Record the unique count for this minute's window
            ip_results.append((current_minute, len(user_set)))
        
        results[ip] = ip_results
    
    return results

2.4 Large-Scale Distributed Processing (Optional) #

If you're dealing with massive datasets (millions/billions of logs), use a distributed framework like Spark. You can leverage built-in window functions to simplify the logic:

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._

// Assume df has columns: minute (int), ip (string), username (string)
val windowSpec = Window.partitionBy("ip")
  .orderBy("minute")
  .rangeBetween(-5, 0)  // 5-minute sliding window (adjust as needed)

val uniqueCountDf = df.withColumn(
  "unique_users_in_window",
  countDistinct("username").over(windowSpec)
)

• Window Boundaries: Be clear whether your window is inclusive/exclusive (e.g., a 5-minute window starting at minute t includes t to t+4, or t to t+5). Consistency here prevents off-by-one errors.
• Memory Management: For real-time processing, regularly clean up empty IP trackers to avoid memory leaks.
• Deduplication: Always deduplicate raw logs first (same minute, IP, username) to reduce unnecessary computation.

内容的提问来源于stack exchange，提问作者O. San