PySpark monotonically_increasing_id()在AWS EMR返回全0的问题求助

问题场景 #

运行以下代码时，本地单实例环境能正常生成唯一递增ID，但部署到AWS EMR后，id列所有值均为0：

from pyspark.sql.functions import monotonically_increasing_id, lit

def foobar(df):
    return (
        df.withColumn("id", monotonically_increasing_id())
        .withColumn("foo", lit("bar"))
        .withColumn("bar", lit("foo"))
    )

somedf = foobar(somedf)
somedf.show() # <-- 所有行id值为0

根据Spark 3.1.3官方文档，monotonically_increasing_id()应生成全局唯一、单调递增的ID，实现逻辑是把分区ID放在高31位，分区内记录偏移量放在低33位，理论上分布式环境也能正常工作。

可能原因 #

1. 单分区+偏移量计算异常：EMR上作业的DataFrame可能被合并为单个分区，且因EMR特定的存储/读取优化（比如S3数据源的分区读取策略、Spark动态分区调整），导致分区内的记录偏移量未正确递增，最终所有ID的高31位（分区ID）和低33位（偏移量）均为0。
2. Spark优化规则干扰：EMR默认启用的部分Spark优化规则（比如逻辑计划合并、谓词下推）可能导致monotonically_increasing_id()的计算时机异常，重复生成初始值。
3. 数据源分区问题：若DataFrame来自S3等对象存储，读取时可能生成空分区或仅单个有效分区，且该分区的记录索引未被正确初始化。

解决方法 #

1. 强制重分区打破单分区限制 #

在生成ID前手动指定分区数，确保存在多个有效分区，这样高31位的分区ID会不同，即使偏移量异常，ID也不会全为0：

def foobar(df):
    return (
        df.repartition(2)  # 根据数据量调整分区数
        .withColumn("id", monotonically_increasing_id())
        .withColumn("foo", lit("bar"))
        .withColumn("bar", lit("foo"))
    )

2. 改用窗口函数生成唯一ID #

如果需要严格的唯一ID（不依赖分区布局），可以用row_number()窗口函数，注意若不需要排序，可指定一个常量排序键（会触发全局 shuffle，数据量大时慎用）：

from pyspark.sql.window import Window

def foobar(df):
    window = Window.orderBy(lit(1))
    return (
        df.withColumn("id", row_number().over(window))
        .withColumn("foo", lit("bar"))
        .withColumn("bar", lit("foo"))
    )

若有可用的排序字段（比如时间戳、主键），替换lit(1)为该字段，性能会更好。

3. 禁用干扰性优化规则 #

若怀疑是Spark优化导致的问题，可在作业启动时添加配置，排除特定优化规则：

# 提交作业时的配置参数
--conf spark.sql.optimizer.excludedRules=org.apache.spark.sql.catalyst.optimizer.ConvertToLocalRelation

4. 固定ID计算结果 #

在生成ID后立即触发一次action（比如cache()或count()），确保ID值被持久化，避免后续操作重新计算导致异常：

def foobar(df):
    temp_df = df.withColumn("id", monotonically_increasing_id()).cache()
    temp_df.count()  # 触发计算，固定ID值
    return temp_df.withColumn("foo", lit("bar")).withColumn("bar", lit("foo"))

补充说明 #

monotonically_increasing_id()的生成依赖物理分区的布局，本身是非确定性的（不同运行可能生成不同ID），如果需要稳定的唯一ID，更推荐使用基于业务字段的哈希ID或自增序列。

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