如何在R中仅向数据仓库(DWH)追加非重复新数据？

核心逻辑 #

要避免重复追加数据，必须通过唯一标识字段（如主键customer_id、记录更新时间戳last_updated）筛选出实时库中DWH表未包含的记录，再执行追加操作。

具体实现方案 #

方案1：基于主键筛选新数据 #

假设表有唯一主键（如customer_id），步骤如下：

1. 从DWH表读取已存在的主键集合
2. 过滤实时数据中已存在的主键记录
3. 仅追加过滤后的新记录

# 1. 读取DWH中已有的customer_id
existing_ids <- dbGetQuery(DWH, "SELECT customer_id FROM public.test_df")$customer_id

# 2. 筛选实时数据中的新记录
new_records <- CUSTOMER[!CUSTOMER$customer_id %in% existing_ids, ]

# 3. 有新记录时才执行追加
if(nrow(new_records) > 0) {
  dbAppendTable(DWH, "test_df", new_records, row.names = FALSE)
}

方案2：基于更新时间戳筛选（支持记录更新场景） #

若实时表有更新时间字段（如last_updated），可同步DWH中没有的、或更新时间更晚的记录：

# 1. 获取DWH表中最新的更新时间
latest_update <- dbGetQuery(DWH, "SELECT MAX(last_updated) FROM public.test_df")[[1]]

# 处理首次同步（DWH表为空）的情况
if(is.na(latest_update)) {
  new_records <- CUSTOMER
} else {
  # 2. 筛选实时数据中更新时间晚于DWH最新时间的记录
  new_records <- CUSTOMER[CUSTOMER$last_updated > latest_update, ]
}

# 3. 追加新记录
if(nrow(new_records) > 0) {
  dbAppendTable(DWH, "test_df", new_records, row.names = FALSE)
}

关键注意事项 #

• dbAppendTable不需要overwrite=TRUE参数，该参数会覆盖整张表，与追加逻辑冲突，直接移除即可。
• 数据量较大时，建议通过数据库端查询过滤（而非全量读取到R内存），减少资源占用：

# 先将实时数据写入数据库临时表
dbWriteTable(DWH, "temp_customer", CUSTOMER, row.names = FALSE, overwrite = TRUE)

# 数据库端直接筛选新记录
new_records <- dbGetQuery(DWH, "
  SELECT * FROM temp_customer 
  WHERE customer_id NOT IN (SELECT customer_id FROM public.test_df)
")

# 追加后删除临时表
dbExecute(DWH, "DROP TABLE temp_customer")

# 执行追加
if(nrow(new_records) > 0) {
  dbAppendTable(DWH, "test_df", new_records, row.names = FALSE)
}

内容的提问来源于stack exchange，提问作者Hassan Elleithy