构建基于Git元数据与OpenTelemetry的Hadoop数据血缘核心库

凌晨三点的告警，把整个数据平台团队都拖入了泥潭。一个核心的Hadoop聚合任务失败，导致下游所有报表数据错乱。日志里充斥着通用的NullPointerException，但失败的根源却难以追溯。这个任务处理的数据来自十几个上游任务，而这些任务的代码在过去一周内有数十次提交。究竟是哪个版本的代码、处理了哪一批数据导致了这次生产事故？我们无从知晓。这种场景在我们的Hadoop集群上反复上演，调试成本高昂，团队疲于奔命。

问题很明确：我们缺乏从代码变更到数据产出的端到端追踪能力。我们需要一个系统，能精确回答“这个HDFS上的数据分区，是由哪个Git Commit、哪个工程师、在什么时间运行的哪个Hadoop任务生成的？”。这就是我们决定自研一个数据血缘核心库的起点。我们的目标不是做一个庞大的平台，而是打造一个轻量级、零侵入、开发者无感的Java核心库。只要任何Hadoop MapReduce项目将其作为依赖，就能自动获得细粒度的可观测性与数据血缘信息。

初步构想与技术选型

最初的构想很简单：在任务启动时，将Git信息作为参数传入，然后记录到某个元数据系统里。但这很快被否决了。手动传递参数容易出错和遗漏，无法保证信息的准确性，也增加了开发者的心智负担。我们需要的是自动化。

方案的核心在于将三件事情联系起来：

1. 代码版本: 通过Git Commit ID唯一标识。
2. 执行上下文: Hadoop任务的ID、名称、输入输出路径等。
3. 可观测性信号: 将上述信息作为分布式追踪的元数据，串联起整个数据处理流程。

围绕这个核心，我们的技术选型决策如下：

• Git元数据注入: 我们选择在构建期自动将Git信息打包进JAR文件。Maven的git-commit-id-plugin或Gradle的类似插件是完美选择。它可以在compile阶段生成一个git.properties文件，包含了commit hash、分支、构建时间等关键信息。应用启动时，只需从classpath加载这个文件即可。
• 可观测性标准: OpenTelemetry是业界的唯一选择。它的标准化API、与主流后端（Jaeger, Prometheus, Zipkin等）的兼容性，以及强大的上下文传播机制，使其成为构建可观测性方案的基石。我们不希望被任何特定厂商绑定。
• 实现载体: 一个独立的Java核心库（hadoop-lineage-core）。所有数据工程团队只需要在他们的pom.xml中添加一个依赖，即可获得所有能力，这最大限度地降低了推广和接入成本。

步骤化实现：从零构建核心库

1. 自动注入Git元数据

这是整个系统的基石。我们在一个独立的pom.xml中配置git-commit-id-plugin，确保任何依赖本库的项目在打包时都能自动嵌入Git信息。

<!-- pom.xml of the hadoop-lineage-core library's user -->
<build>
    <plugins>
        <plugin>
            <groupId>pl.project13.maven</groupId>
            <artifactId>git-commit-id-plugin</artifactId>
            <version>4.9.10</version>
            <executions>
                <execution>
                    <id>get-the-git-infos</id>
                    <goals>
                        <goal>revision</goal>
                    </goals>
                    <phase>initialize</phase>
                </execution>
            </executions>
            <configuration>
                <dotGitDirectory>${project.basedir}/.git</dotGitDirectory>
                <prefix>git</prefix>
                <verbose>false</verbose>
                <generateGitPropertiesFile>true</generateGitPropertiesFile>
                <generateGitPropertiesFilename>${project.build.outputDirectory}/git.properties</generateGitPropertiesFilename>
                <format>properties</format>
                <commitIdGenerationMode>full</commitIdGenerationMode>
            </configuration>
        </plugin>
    </plugins>
</build>

这个配置会在编译后的classes目录下生成一个git.properties文件。接下来，我们需要一个工具类来加载这些信息。

package com.mycorp.hadoop.lineage.util;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.InputStream;
import java.util.Properties;

/**
 * 负责加载和提供编译时注入的Git元数据。
 * 在真实项目中，这里应该使用单例模式，避免重复加载。
 */
public final class GitMetadata {

    private static final Logger logger = LoggerFactory.getLogger(GitMetadata.class);
    private static final Properties properties = new Properties();
    private static final String UNKNOWN = "unknown";

    static {
        try (InputStream stream = GitMetadata.class.getClassLoader().getResourceAsStream("git.properties")) {
            if (stream != null) {
                properties.load(stream);
            } else {
                logger.warn("git.properties not found on classpath. Git metadata will be unavailable.");
            }
        } catch (Exception e) {
            logger.error("Failed to load git.properties", e);
        }
    }

    public static String getCommitId() {
        return properties.getProperty("git.commit.id.full", UNKNOWN);
    }

    public static String getBranch() {
        return properties.getProperty("git.branch", UNKNOWN);
    }

    public static String getBuildTime() {
        return properties.getProperty("git.build.time", UNKNOWN);
    }

    public static String getBuildUser() {
        return properties.getProperty("git.build.user.name", UNKNOWN);
    }
    
    // 禁止实例化
    private GitMetadata() {}
}

2. 初始化OpenTelemetry SDK

我们需要一个中心化的配置和初始化逻辑，用于创建TracerProvider。在真实项目中，这些配置（如OTLP exporter的endpoint）应该来自外部配置文件，而不是硬编码。

package com.mycorp.hadoop.lineage.opentelemetry;

import com.mycorp.hadoop.lineage.util.GitMetadata;
import io.opentelemetry.api.OpenTelemetry;
import io.opentelemetry.api.trace.Tracer;
import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporter;
import io.opentelemetry.sdk.OpenTelemetrySdk;
import io.opentelemetry.sdk.resources.Resource;
import io.opentelemetry.sdk.trace.SdkTracerProvider;
import io.opentelemetry.sdk.trace.export.BatchSpanProcessor;
import io.opentelemetry.semconv.resource.attributes.ResourceAttributes;

import java.util.concurrent.TimeUnit;

/**
 * OpenTelemetry SDK 的封装与初始化器。
 * 管理 TracerProvider 的生命周期。
 */
public final class OtelSdkProvider {

    private static volatile OpenTelemetry openTelemetryInstance;
    private static final String INSTRUMENTATION_NAME = "com.mycorp.hadoop.lineage";
    private static final String INSTRUMENTATION_VERSION = "1.0.0";

    // 使用双重检查锁定确保线程安全的单例
    public static OpenTelemetry getOpenTelemetry() {
        if (openTelemetryInstance == null) {
            synchronized (OtelSdkProvider.class) {
                if (openTelemetryInstance == null) {
                    openTelemetryInstance = initializeSdk();
                }
            }
        }
        return openTelemetryInstance;
    }
    
    public static Tracer getTracer() {
        return getOpenTelemetry().getTracer(INSTRUMENTATION_NAME, INSTRUMENTATION_VERSION);
    }

    private static OpenTelemetry initializeSdk() {
        // 1. 配置 OTLP Exporter，指向你的 Collector 地址
        // 在生产环境中，这个 endpoint 必须是可配置的
        OtlpGrpcSpanExporter spanExporter = OtlpGrpcSpanExporter.builder()
                .setEndpoint("http://your-otel-collector:4317")
                .setTimeout(2, TimeUnit.SECONDS)
                .build();

        // 2. 创建资源，包含服务名和我们注入的Git元数据
        // 这些信息会作为所有 Span 的公共属性
        Resource resource = Resource.getDefault()
                .merge(Resource.create(io.opentelemetry.api.common.Attributes.builder()
                        .put(ResourceAttributes.SERVICE_NAME, "hadoop-job-processor")
                        .put("git.commit.id", GitMetadata.getCommitId())
                        .put("git.branch", GitMetadata.getBranch())
                        .put("git.build.time", GitMetadata.getBuildTime())
                        .build()));

        // 3. 构建 TracerProvider，并连接 Exporter
        SdkTracerProvider sdkTracerProvider = SdkTracerProvider.builder()
                .addSpanProcessor(BatchSpanProcessor.builder(spanExporter).build())
                .setResource(resource)
                .build();
        
        // 4. 构建并注册全局的 OpenTelemetry 实例
        OpenTelemetrySdk sdk = OpenTelemetrySdk.builder()
                .setTracerProvider(sdkTracerProvider)
                .buildAndRegisterGlobal();

        // 5. 增加 JVM 关闭钩子，确保在程序退出时能刷新并关闭 SDK，避免数据丢失
        Runtime.getRuntime().addShutdownHook(new Thread(sdkTracerProvider::close));

        return sdk;
    }
    
    private OtelSdkProvider() {}
}

3. 核心挑战：跨进程的上下文传播

这是最棘手的部分。Hadoop MapReduce是一个分布式计算模型。Driver进程负责配置和提交Job，而Map和Reduce任务则运行在集群中完全不同的JVM里。OpenTelemetry的上下文（Context）默认是存储在ThreadLocal中的，它无法跨越进程边界。

我们的解决方案是：

1. 在Driver端: 创建一个根Span，并将当前的Trace Context序列化为一个字符串。
2. 注入配置: 将这个序列化后的字符串存入Hadoop的Configuration对象中。这个对象会被Hadoop框架分发到所有的Map和Reduce任务节点。
3. 在Task端: 在Map和Reduce任务的setup()方法（任务初始化时调用）中，从Configuration中读取这个字符串。
4. 恢复上下文: 将字符串反序列化为Trace Context，并将其设置为当前线程的上下文。这样，在Task内部创建的任何新Span都会自动成为Driver端根Span的子Span。

OpenTelemetry提供了TextMapPropagator接口来帮助我们完成序列化和反序列化。

sequenceDiagram
    participant Driver as Hadoop Job Driver (Client JVM)
    participant HadoopFramework as Hadoop Framework (YARN)
    participant Mapper as Mapper Task (Node JVM)
    participant Reducer as Reducer Task (Node JVM)

    Driver->>Driver: 1. OtelSdkProvider.getTracer().spanBuilder("wordcount-job").startSpan()
    Note right of Driver: 创建根 Span

    Driver->>Driver: 2. 将 Trace Context 注入 Hadoop Configuration
    Driver->>HadoopFramework: 3. job.submit() 提交带有上下文的配置

    HadoopFramework->>Mapper: 4. 分发配置并启动 Mapper
    Mapper->>Mapper: 5. setup() 方法从 Configuration 中提取上下文
    Mapper->>Mapper: 6. 恢复 Trace Context, 创建 Mapper Span (成为根Span的子Span)
    Mapper->>Mapper: 7. 执行 map() 逻辑...
    Mapper->>Mapper: 8. cleanup(), 结束 Mapper Span

    HadoopFramework->>Reducer: 9. 分发配置并启动 Reducer
    Reducer->>Reducer: 10. setup() 方法从 Configuration 中提取上下文
    Reducer->>Reducer: 11. 恢复 Trace Context, 创建 Reducer Span
    Reducer->>Reducer: 12. 执行 reduce() 逻辑...
    Reducer->>Reducer: 13. cleanup(), 结束 Reducer Span

    Reducer->>Driver: (Task completion signals)
    Mapper->>Driver: (Task completion signals)

    Driver->>Driver: 14. 任务结束, 结束根 Span

下面是实现这个传播机制的核心工具类。

package com.mycorp.hadoop.lineage.opentelemetry;

import io.opentelemetry.api.trace.Span;
import io.opentelemetry.api.trace.StatusCode;
import io.opentelemetry.context.Context;
import io.opentelemetry.context.Scope;
import io.opentelemetry.context.propagation.TextMapGetter;
import io.opentelemetry.context.propagation.TextMapSetter;
import org.apache.hadoop.conf.Configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nullable;
import java.util.HashMap;
import java.util.Map;

public final class HadoopContextPropagator {

    private static final Logger logger = LoggerFactory.getLogger(HadoopContextPropagator.class);
    private static final String CONTEXT_PREFIX = "otel.propagation.";

    // Setter 用于将上下文注入 Configuration
    private static final TextMapSetter<Configuration> setter =
            (carrier, key, value) -> {
                if (carrier != null) {
                    carrier.set(CONTEXT_PREFIX + key, value);
                }
            };

    // Getter 用于从 Configuration 中提取上下文
    private static final TextMapGetter<Configuration> getter = new TextMapGetter<>() {
        @Override
        public Iterable<String> keys(Configuration carrier) {
            Map<String, String> prefixed = carrier.getPropsWithPrefix(CONTEXT_PREFIX);
            return prefixed.keySet();
        }

        @Nullable
        @Override
        public String get(@Nullable Configuration carrier, String key) {
            if (carrier == null) {
                return null;
            }
            return carrier.get(CONTEXT_PREFIX + key);
        }
    };
    
    /**
     * 在 Driver 端调用，将当前 OTel 上下文注入 Hadoop Configuration。
     * @param conf Hadoop 配置对象
     */
    public static void injectContext(Configuration conf) {
        if (conf == null) {
            logger.warn("Configuration is null, cannot inject OpenTelemetry context.");
            return;
        }
        try {
            OtelSdkProvider.getOpenTelemetry().getPropagators().getTextMapPropagator()
                .inject(Context.current(), conf, setter);
            logger.info("Successfully injected OpenTelemetry context into Hadoop Configuration.");
        } catch (Exception e) {
            logger.error("Failed to inject OpenTelemetry context.", e);
        }
    }
    
    /**
     * 在 Mapper/Reducer 的 setup() 方法中调用，从 Configuration 中提取并激活上下文。
     * @param conf 从 TaskAttemptContext 获取的配置对象
     * @return 返回提取的上下文，用于后续创建子 Span。
     */
    public static Context extractContext(Configuration conf) {
        if (conf == null) {
            logger.warn("Configuration is null, cannot extract OpenTelemetry context. Returning empty context.");
            return Context.root();
        }
        try {
            return OtelSdkProvider.getOpenTelemetry().getPropagators().getTextMapPropagator()
                .extract(Context.current(), conf, getter);
        } catch (Exception e) {
            logger.error("Failed to extract OpenTelemetry context. Returning empty context.", e);
            return Context.root();
        }
    }
    
    // 一个辅助方法，封装了 Span 的标准创建和结束逻辑，包含错误处理
    public static void executeWithSpan(String spanName, Context parentContext, Runnable task) {
        Span span = OtelSdkProvider.getTracer().spanBuilder(spanName)
                .setParent(parentContext)
                .startSpan();
        try (Scope scope = span.makeCurrent()) {
            task.run();
            span.setStatus(StatusCode.OK);
        } catch (Exception e) {
            span.setStatus(StatusCode.ERROR, e.getMessage());
            span.recordException(e);
            throw e;
        } finally {
            span.end();
        }
    }

    private HadoopContextPropagator() {}
}

4. 应用到实际的MapReduce任务

现在，我们把所有零件组装起来。我们将修改经典的WordCount示例来展示这个库的用法。

Driver 端 (main方法):

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import com.mycorp.hadoop.lineage.opentelemetry.OtelSdkProvider;
import com.mycorp.hadoop.lineage.opentelemetry.HadoopContextPropagator;
import io.opentelemetry.api.trace.Span;
import io.opentelemetry.api.trace.StatusCode;
import io.opentelemetry.context.Scope;

public class InstrumentedWordCount {

    public static void main(String[] args) throws Exception {
        // 1. 初始化 OTel SDK
        Span rootSpan = OtelSdkProvider.getTracer().spanBuilder("wordcount-job-driver").startSpan();
        
        // 2. 将根 Span 设为当前上下文
        try (Scope scope = rootSpan.makeCurrent()) {
            Configuration conf = new Configuration();
            
            // 3. 关键步骤：将当前上下文注入到 Hadoop Configuration
            HadoopContextPropagator.injectContext(conf);

            // 添加业务属性和血缘信息到根 Span
            rootSpan.setAttribute("hadoop.job.name", "WordCount");
            rootSpan.setAttribute("data.input.path", args[0]);
            rootSpan.setAttribute("data.output.path", args[1]);

            Job job = Job.getInstance(conf, "Instrumented WordCount");
            job.setJarByClass(InstrumentedWordCount.class);
            job.setMapperClass(TokenizerMapper.class);
            job.setCombinerClass(IntSumReducer.class);
            job.setReducerClass(IntSumReducer.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);
            FileInputFormat.addInputPath(job, new Path(args[0]));
            FileOutputFormat.setOutputPath(job, new Path(args[1]));
            
            boolean success = job.waitForCompletion(true);
            
            if (success) {
                rootSpan.setStatus(StatusCode.OK, "Job completed successfully");
            } else {
                rootSpan.setStatus(StatusCode.ERROR, "Job failed");
            }
            System.exit(success ? 0 : 1);

        } catch (Exception e) {
            rootSpan.setStatus(StatusCode.ERROR, "Job execution failed with exception");
            rootSpan.recordException(e);
            throw e;
        } finally {
            // 4. 确保根 Span 被关闭
            rootSpan.end();
            // 在实际应用中，你可能需要等待一段时间确保span被导出
            Thread.sleep(5000); 
        }
    }
}

Mapper 端:

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import com.mycorp.hadoop.lineage.opentelemetry.HadoopContextPropagator;
import com.mycorp.hadoop.lineage.opentelemetry.OtelSdkProvider;
import io.opentelemetry.api.trace.Span;
import io.opentelemetry.context.Context;
import io.opentelemetry.context.Scope;

import java.io.IOException;
import java.util.StringTokenizer;

public class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {

    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();
    private Context parentContext;

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        // 1. 在任务初始化时，从 Configuration 中提取父上下文
        this.parentContext = HadoopContextPropagator.extractContext(context.getConfiguration());
    }

    @Override
    public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
        // 2. 为每一次 map 调用创建一个子 Span
        Span mapSpan = OtelSdkProvider.getTracer().spanBuilder("wordcount-mapper")
                .setParent(this.parentContext)
                .startSpan();
        
        try (Scope scope = mapSpan.makeCurrent()) {
            mapSpan.setAttribute("map.input.key", key.toString());
            mapSpan.setAttribute("map.input.value.length", value.toString().length());
            
            StringTokenizer itr = new StringTokenizer(value.toString());
            while (itr.hasMoreTokens()) {
                word.set(itr.nextToken());
                context.write(word, one);
            }
            
        } finally {
            mapSpan.end();
        }
    }
}

Reducer的改造与Mapper类似，在setup()中提取上下文，在reduce()中创建Span。

最终成果与价值

部署这套核心库后，我们的数据平台发生了质的变化。每当一个Hadoop任务运行时，一个完整的分布式追踪链就会被发送到我们的可观测性后端（如Jaeger）。

这个追踪链不仅包含了Driver, Mapper, Reducer的执行耗时、依赖关系等性能指标，更重要的是，每个Span都带有我们注入的Git元数据。现在，当数据出现问题时，我们的排查流程是：

1. 根据数据产出路径或时间，在Jaeger中搜索相关的wordcount-job-driver Span。
2. 打开Trace，立刻就能在Resource属性中看到生成这份数据的git.commit.id、git.branch和构建者。
3. 点击该Commit ID，可以直接跳转到Git仓库，进行代码CR，分析变更逻辑。
4. Trace视图清晰地展示了整个任务的生命周期，哪个阶段耗时最长，哪个Mapper处理了异常数据，一目了然。

这种从数据到代码的直接链接，将问题排查时间从数小时甚至数天，缩短到了几分钟。它真正实现了我们最初的目标：为每一个数据产出都打上一个不可变的、可追溯的“身份烙印”。

局限性与未来迭代

这个核心库虽然解决了MapReduce的血缘和可观测性问题，但它并非万能。

首先，当前实现强依赖于Hadoop MapReduce的编程模型。对于Spark、Flink等其他计算引擎，上下文传播的机制完全不同。例如，对于Spark，我们可能需要利用SparkContext.addSparkListener来监听任务生命周期，并通过闭包将上下文传递到Executor的Task中，实现起来更为复杂。

其次，我们的方案只解决了“代码到数据”的血缘。一个完整的数据治理体系还需要“数据到数据”的血缘，即一个任务的输出作为另一个任务的输入。这需要解析任务的输入输出路径，并将其作为Span的Link关联起来，形成一个跨任务的DAG图。这会是我们下一个迭代版本的核心功能。

最后，对于由Airflow、Oozie等工作流调度系统触发的任务，追踪上下文应该从调度器就开始，并一路透传到Hadoop任务内部。这需要对调度器进行插件式改造，以支持W3C Trace Context标准的传播。这会是一个更宏大的工程，但也是构建统一数据平台可观测性的必经之路。