Pyspark如何定期读取增量Hive表而非每次全量扫描？

首先不推荐你用rowId/rownum标记增量的思路：Hive本身没有内置固定的行ID，rownum是SQL查询时动态计算生成的，只要原表有数据删除、插入或者分区调整，同一行数据的rownum会发生变化，无法稳定作为增量边界的标记，容易出现漏数或者重复读取的问题。

下面是几个适合离线批处理场景、新手友好的实现方案，按推荐优先级排序：

方案1：按时间字段增量同步（最推荐） #

适用前提 #

原Hive表带有数据写入时间字段，比如create_time、upload_time，或者是分区表有dt这类时间分区字段。

实现步骤 #

• 提前新建一张同步标记表，存储任务的同步边界，表结构参考：
  • task_name：同步任务名称，用来区分不同任务
  • last_sync_value：上次同步的最大时间值
  • update_time：标记的更新时间
• 每次任务启动时，先从标记表读取上次同步的最大时间
• 读取原表时添加时间过滤条件，只读取大于上次同步时间的新数据
• 完成聚合转换、写入目标表后，将本次处理的原表最大时间更新到标记表

优缺点 #

• 优点：逻辑简单易懂，出错概率低，就算有延迟数据也可以通过调整时间范围补数
• 缺点：需要原表有可用的时间字段，如果有历史数据更新需要额外处理

方案2：按分区增量同步 #

适用前提 #

原Hive表是按时间分区的，比如按天分区dt='yyyy-MM-dd'、按小时分区hour='HH'

实现步骤 #

• 同步标记表存储上次同步的最大分区值
• 每次任务启动时，先查询原表的所有分区，筛选出大于上次同步分区的新增分区
• 只读取这些新增分区的数据做处理
• 处理完成后将最新的分区值更新到标记表

优缺点 #

• 优点：直接利用Hive分区剪枝特性，不需要扫描全表数据，读取性能极高
• 缺点：如果有历史分区被重写更新，默认逻辑无法识别，需要额外加分区修改时间的判断逻辑

方案3：按业务主键增量同步 #

适用前提 #

原表没有时间字段也没有分区，但存在全局唯一且自增的业务主键，同时原表只有新增数据，不会有历史数据的修改和删除

实现步骤 #

• 同步标记表存储上次同步的最大主键值
• 每次读取原表时添加过滤条件where 主键字段 > 上次最大主键值
• 处理完成后更新本次处理的最大主键值到标记表

优缺点 #

• 优点：不依赖时间字段和分区，实现简单
• 缺点：无法识别历史数据的更新，适用场景非常有限

可直接复用的代码示例 #

from pyspark.sql import functions as F

# 任务配置
TASK_NAME = "user_order_agg_task"
SOURCE_HIVE_TABLE = "ods.ods_user_order"
TARGET_HIVE_TABLE = "dws.dws_user_order_stat"
MARK_TABLE = "common.task_sync_mark"

# 读取上次同步边界
last_sync_df = spark.sql(f"""
    select last_sync_value from {MARK_TABLE} where task_name = '{TASK_NAME}'
""")
# 首次运行默认同步全量历史数据
last_sync_time = last_sync_df.collect()[0][0] if last_sync_df.count() > 0 else "1970-01-01 00:00:00"

# 增量读取原表数据
source_df = spark.sql(f"""
    select * from {SOURCE_HIVE_TABLE} where create_time > '{last_sync_time}'
""")

# 你的业务聚合转换逻辑
agg_result_df = source_df.groupBy("dt", "user_id").agg(
    F.countDistinct("order_id").alias("order_cnt"),
    F.sum("pay_amount").alias("total_pay")
)

# 追加写入目标表
agg_result_df.write.mode("append").insertInto(TARGET_HIVE_TABLE)

# 更新同步标记
current_max_time = source_df.agg(F.max("create_time")).collect()[0][0]
spark.sql(f"""
    insert overwrite table {MARK_TABLE}
    select * from {MARK_TABLE} where task_name != '{TASK_NAME}'
    union all
    select '{TASK_NAME}' as task_name, '{current_max_time}' as last_sync_value, current_timestamp() as update_time
""")

内容的提问来源于stack exchange，提问作者Albin Chandy