Azure Data Explorer(ADX)重复数据处理的更多可行方案咨询

1. 物化视图自动维护唯一数据集 #

利用物化视图的持续聚合能力，自动保留每个唯一键组合的最新记录，无需手动执行定期清理任务，适合需要实时保持数据唯一性的场景。

示例配置：

.create materialized view with (backfill=true) MV_UniqueRecords on table MySourceTable
{
    MySourceTable
    | summarize arg_max(ingestion_time(), *) by UniqueKeyColumn1, UniqueKeyColumn2
}

• 核心逻辑：通过arg_max函数，针对指定的唯一键组合，始终保留最新写入的记录，物化视图会持续同步源表数据并自动去重。
• 优势：无需手动调度，实时维护唯一数据集；查询时直接访问物化视图，性能优于直接查询源表后去重。

2. 批量更新策略标记+清理重复项 #

通过配置更新策略自动标记重复数据，再结合周期性清理操作，适合需要保留源表原始结构、且重复数据周期性产生的场景。

步骤示例：

1. 创建用于标记重复项的辅助表：

.create table DuplicateFlags (RecordId: string, IsDuplicate: bool)

2. 给源表配置更新策略，自动标记重复记录：

.alter table MySourceTable policy update @'[{"IsEnabled": true, "Source": "MySourceTable", "Query": "MySourceTable | summarize count() by RecordId | where count_ > 1 | project RecordId, IsDuplicate=true", "IsTransactional": false, "PropagateIngestionProperties": false}]'

3. 定期执行清理（可结合自动化工具调度）：

let duplicateIds = DuplicateFlags | where IsDuplicate == true | project RecordId;
MySourceTable
| where RecordId in (duplicateIds)
| partition by RecordId (
    sort by ingestion_time() desc
    | skip 1
    | delete from MySourceTable where RecordId == RecordId
)

3. 流引入阶段实时过滤重复 #

如果使用流引入方式接收数据，可在数据落地ADX前直接加入去重逻辑，从源头阻断重复数据进入存储。

示例流引入查询：

.ingest into table RealTimeData (
    MyStreamSource
    | summarize arg_max(EventTimestamp, *) by EventId
)

• 核心逻辑：对每一批流数据，按业务唯一键（如EventId）筛选出最新的记录，仅将非重复数据写入目标表。
• 适用场景：IoT设备日志、实时业务事件等高吞吐量流数据场景。

4. 分区级重复清理优化 #

针对大表场景，按分区（如按日期分区）执行去重操作，降低单批次处理的数据量，提升清理效率。

示例分区去重查询：

MyPartitionedTable
| where ingestion_time() between (startofday(ago(7d)) .. startofday(ago(1d)))
| summarize arg_max(ingestion_time(), *) by UniqueKey
| into Temp_CleanedPartition
.drop table MyPartitionedTable partition (datetime(2024-05-01))
.ingest into table MyPartitionedTable partition (datetime(2024-05-01)) from Temp_CleanedPartition
.drop table Temp_CleanedPartition

• 优势：避免全表扫描，仅针对目标分区处理，适合数据量较大、按时间或业务维度分区的表。

内容的提问来源于stack exchange，提问作者NIKHIL A V