如何在Databricks中捕获Delta表的插入/更新行数

直接从DML操作结果获取 #

执行MERGE、INSERT或UPDATE后，可直接通过操作返回对象的num_affected_rows属性拿到数值：

# 以MERGE操作为例
merge_result = spark.table("target_table").merge(
    spark.table("source_table"),
    "target_table.id = source_table.id"
).whenMatchedUpdateAll().whenNotMatchedInsertAll().execute()

# 直接提取数值
affected_rows = merge_result.num_affected_rows
print(affected_rows)  # 输出31

从日志文本中匹配提取 #

如果只有包含num_affected_rows=31的日志内容，用正则表达式提取数字：

import re

log_content = "num_affected_rows=31"
matched_num = re.findall(r'num_affected_rows=(\d+)', log_content)[0]
print(int(matched_num))  # 输出31

Delta表通过历史记录查询 #

若是Delta格式的表，可查询表的最新操作历史获取受影响行数：

-- 查询最近1条操作记录
DESCRIBE HISTORY delta.`/your/table/path` LIMIT 1;

结果的operationMetrics字段中，会包含numTargetRowsInserted、numTargetRowsUpdated等细分指标，按需提取求和即可。

内容的提问来源于stack exchange，提问作者bigdata techie