如何在Hive UDF中获取taskID、mapperID（类似Spark的partitionID）

Hive底层依赖的计算引擎（MapReduce/Tez/Spark）都会将任务标识写入运行时上下文，你可以通过以下两种常用方式在UDF中读取，效果等价于Spark的partitionID：

方法1：读取JVM系统属性（最简实现） #

不需要重写UDF的额外方法，直接在evaluate方法或初始化逻辑中读取系统参数即可，兼容绝大多数Hadoop版本和计算引擎：

• MapReduce引擎参数：mapreduce.task.attempt.id（新版本）、mapred.task.id（旧版本兼容）
• Tez引擎参数：tez.task.id
• Spark引擎参数：spark.task.partitionId

代码示例：

public class GetTaskIdUDF extends UDF {
    private Integer taskId = null;

    public IntWritable evaluate() {
        if (taskId == null) {
            // 兼容多引擎依次读取
            String idStr = System.getProperty("mapreduce.task.attempt.id");
            if (idStr == null) idStr = System.getProperty("tez.task.id");
            if (idStr == null) idStr = System.getProperty("spark.task.partitionId");
            
            // 提取纯数字任务ID
            if (idStr != null) {
                String[] parts = idStr.split("_");
                // MapReduce/Tez的ID格式为attempt_xxx_xxx_m_000001_0，第5段为序号
                taskId = Integer.parseInt(parts.length >= 5 ? parts[4] : parts[0]);
            } else {
                taskId = -1;
            }
        }
        return new IntWritable(taskId);
    }
}

方法2：通过configure方法读取Hadoop配置（更稳定） #

如果你的UDF继承了Hive官方的org.apache.hadoop.hive.ql.exec.UDF类，可以重写configure方法直接获取JobConf对象，读取参数更稳定，不会受JVM系统属性覆盖的影响：
代码示例：

public class GetTaskIdUDF extends UDF {
    private int taskId;

    @Override
    public void configure(JobConf job) {
        String attemptId = job.get("mapreduce.task.attempt.id", job.get("mapred.task.id", "-1"));
        String[] parts = attemptId.split("_");
        taskId = Integer.parseInt(parts.length >=5 ? parts[4] : "-1");
    }

    public IntWritable evaluate() {
        return new IntWritable(taskId);
    }
}

注意事项 #

• 不要在UDF的静态代码块中读取上述参数，静态类加载时任务上下文还未初始化，会读取到空值
• 同一个Task进程内的所有UDF调用获取到的ID一致，不同Task的ID唯一，可直接用作分区标识使用

内容的提问来源于stack exchange，提问作者Shawn.X