1. Log Deduplication #

1.1 Why Deduplicate Logs? #

Duplicate logs are a common problem in real-time log processing, primarily caused by the following factors:

• Duplicate requests from client-side retry mechanisms
• Duplicate message delivery due to network jitter
• Improper handling of idempotency in distributed systems
• Duplicate data from log collection pipelines

Value of Deduplication:

• Save on storage costs (reduces storage space by 30%-50%)
• Improve analysis accuracy (avoid duplicate counting)
• Optimize downstream processing performance

1.2 Understanding Flink SQL Deduplication Principles #

Core mechanism:

• PARTITION BY: Defines the deduplication dimension (e.g., reqid)
• ROW_NUMBER() window function: Assigns a unique sequence number to each record within a partition
• ORDER BY: Determines the retention policy (ASC retains the first record, DESC retains the last one)
• row_num = 1: Filters to retain the target record

2. Environment Preparation and Basic Configuration #

2.1 Prerequisites #

2.2 Log Source Table Configuration (Kafka) #

CREATE TABLE raw_logs (
  log_time TIMESTAMP(3),
  reqid    VARCHAR(40),    -- Request ID (deduplication key)
  service  VARCHAR(20),    -- Service name
  level    VARCHAR(10),    -- Log level
  message  STRING,         -- Log content
  proc_time AS PROCTIME(), -- Processing time
  
  -- Event time configuration (Optional)
  WATERMARK FOR log_time AS log_time - INTERVAL '10' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'raw_logs',
  'properties.bootstrap.servers' = 'kafka:9092',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json',
  'json.ignore-parse-errors' = 'true'
);

3. Hands-On Deduplication: From Simple to Complex Scenarios #

3.1 Basic Deduplication: Deduplicate by reqid (Keep the Latest Log) #

CREATE TABLE dedup_logs_simple (
  reqid    VARCHAR(40) PRIMARY KEY NOT ENFORCED,
  log_time TIMESTAMP(3),
  service  VARCHAR(20),
  message  STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'dedup_logs',
  'properties.bootstrap.servers' = 'kafka:9092',
  'format' = 'json'
);

INSERT INTO dedup_logs_simple
SELECT reqid, log_time, service, message
FROM (
  SELECT *,
    ROW_NUMBER() OVER (
      PARTITION BY reqid
      ORDER BY proc_time DESC    -- Keep latest by processing time
    ) AS row_num
  FROM raw_logs
) 
WHERE row_num = 1;

3.2 Composite Key Deduplication: reqid + Service Name #

INSERT INTO dedup_logs_simple
SELECT reqid, log_time, service, message
FROM (
  SELECT *,
    ROW_NUMBER() OVER (
      PARTITION BY reqid, service  -- Partition by composite key
      ORDER BY proc_time DESC       -- Keep latest by processing time
    ) AS row_num
  FROM raw_logs
) 
WHERE row_num = 1;

3.3 Time Window Deduplication (Keep the First Entry per Hour) #

INSERT INTO dedup_logs_simple
SELECT reqid, log_time, service, message
FROM (
  SELECT *,
    ROW_NUMBER() OVER (
      PARTITION BY reqid, 
                   HOUR(proc_time)  -- Partition by hour
      ORDER BY proc_time ASC        -- Keep the first record per hour
    ) AS row_num
  FROM raw_logs
) 
WHERE row_num = 1;

4. Advanced Configuration and Performance Optimization #

4.1 State Management (State TTL Explained) #

The behavior of TTL in different modes is as follows:

4.2 Processing Time vs. Event Time #

Complete Event Time Example:

CREATE TABLE dedup_event_time (
  reqid    VARCHAR(40),
  log_time TIMESTAMP(3),
  service  VARCHAR(20),
  message  STRING
) WITH (/* Kafka configuration */);

INSERT INTO dedup_event_time
SELECT reqid, log_time, service, message
FROM (
  SELECT *,
    ROW_NUMBER() OVER (
      PARTITION BY reqid
      ORDER BY log_time DESC    -- Sort by event time
    ) AS row_num
  FROM raw_logs
  WHERE WATERMARK(log_time) IS NOT NULL -- Ensure event time is valid
) 
WHERE row_num = 1;

4.3 Performance Optimization Techniques #

1. Partition Key Optimization:

-- Add a prefix to reduce state size (for high-cardinality keys)
PARTITION BY SUBSTRING(reqid, 1, 4), reqid

2. Resource Tuning:

3. MiniBatch Optimization (see MiniBatch Configuration):

SET 'table.exec.mini-batch.enabled' = 'true';
SET 'table.exec.mini-batch.size' = '5000';

5. Best Practices for Production Environments #

5.1 End-to-End Exactly-Once Semantics #

After deduplicating data in Flink, use Flink's exactly-once semantics to ensure data accuracy.

5.2 Troubleshooting Guide for Common Issues #

#

#