企业官网建设流程全解析

• 大数据
• 批处理
• 流处理
• 数据工程

本文围绕 Apache Beam 仓库中 .test-infra/jupyter 目录下的测试指标分析工作流展开，讲解如何在本地用 Jupyter Notebook 从 Jenkins（ci-beam.apache.org）拉取 PreCommit 任务与单个测试用例的统计数据，并通过 pandas、matplotlib 完成耗时画像与分位数分析。读完本文，你将掌握这套测试基建的具体搭建步骤、Jenkins API 的调用约束、Notebook 各代码单元的解析逻辑，以及向该目录提交改动时应遵守的约定。

一、目录定位：测试指标从采集到分析的第一站

Apache Beam 的持续集成体系分为 GitHub Actions 与历史遗留的 Jenkins 两大部分，具体分工可参见仓库根目录的 CI.md：绝大多数 CI 工作流已迁移到 GitHub Actions，而.test-infra目录则沉淀了与测试基础设施相关的脚本、指标同步与集群编排设施。

其中 .test-infra/jupyter/README.md 明确说明：该目录存放用于采集与分析测试指标（test metrics）的 Jupyter Notebook。当前目录中实际只有一个 Notebook：precommit_job_times.ipynb，它对应目录内一句话定位——"This notebook fetches test statistics from Jenkins"，即从 Jenkins 抓取 PreCommit 任务的统计信息。它并非通用数据分析教程，而是 Beam 团队维护的一套可复用的 CI 指标观测工具，用于回答诸如"Java/Python/Go 的 PreCommit 任务最近跑得有多慢""哪些单个测试用例耗时最长"这类问题。

二、环境准备：基于 pip + venv 的 Jupyter 安装

README 给出了面向 Linux 的官方安装步骤，核心是使用 Python 虚拟环境隔离依赖：

python3 -m venv ~/virtualenvs/jupyter source ~/virtualenvs/jupyter/bin/activate pip install jupyter # Optional packages, for example: pip install pandas matplotlib requests cd .test-infra/jupyter jupyter notebook # Should open a browser window.

要点说明：

• python3 -m venv ~/virtualenvs/jupyter创建独立虚拟环境，避免污染系统 Python；
• source .../bin/activate激活环境后，pip install jupyter安装 Notebook 服务本体；
• pandas、matplotlib、requests 属于"可选但实际必需"的依赖——Notebook 的第一个代码单元直接import pandas as pd / numpy / matplotlib / requests，缺一不可；
• 启动前先cd .test-infra/jupyter，这样jupyter notebook打开后能直接浏览到precommit_job_times.ipynb。

三、Notebook 总览：四个层次的 Jenkins 指标分析

precommit_job_times.ipynb（nbformat 4，Python 3 内核）按执行顺序组织为六个代码单元，整体形成"任务级耗时采集 → 时间窗过滤 → 可视化 → 分位数统计 → 用例级数据采集 → 交互式分析"的完整链路：

四、数据源约束：访问 ci-beam.apache.org 的 API 红线

Notebook 的 Markdown 说明里有一段必须遵守的硬性约束：

Note:Requests toci-beam.apache.orgmust contain a ?depth= or ?tree= argument, otherwise your IP will get banned. Policy

翻译过来即：所有发往ci-beam.apache.org的请求必须携带?depth=或?tree=参数，否则 IP 会被封禁。这一策略来自 ASF 的 Jenkins API 使用规范，目的是防止未限制返回深度的请求拖垮 Jenkins 实例。因此 Notebook 中每一处requests.get都显式携带了tree或depth参数——这是该仓库代码中体现 API 红线最直接的地方，后续任何新增采集逻辑都应沿用这一约定。

仓库内另一处 Jenkins 数据管道 .test-infra/metrics/sync/jenkins/syncjenkins.py 也遵循同样的约束，例如其fetchJobs()使用'https://ci-beam.apache.org/api/json?tree=jobs[name,url,lastCompletedBuild[id]]&depth=1'，将 Jenkins 构建记录同步到 PostgreSQL 的jenkins_builds表（含timing_queuingDurationMillis、timing_totalDurationMillis等与 Notebook 同名概念的时间字段），可见"任务耗时"是整个 Beam 测试指标体系共享的核心观测维度。

五、任务级数据采集：Build 类与 TimeInQueueAction

单元 2 是整个 Notebook 的数据入口，首先定义了一个继承自dict的Build类，把 Jenkins 构建 JSON 规整为 DataFrame 可直接消费的字段：

# Fetch precommit job data from Jenkins. class Build(dict): def __init__(self, job_name, json): self['job_name'] = job_name self['result'] = json['result'] self['number'] = json['number'] self['timestamp'] = pd.Timestamp.utcfromtimestamp(json['timestamp'] / 1000) self['queuingDurationMillis'] = -1 self['totalDurationMillis'] = -1 for action in json['actions']: if action.get('_class', None) == 'jenkins.metrics.impl.TimeInQueueAction': self['queuingDurationMinutes'] = action['queuingDurationMillis'] / 60000. self['totalDurationMinutes'] = action['totalDurationMillis'] / 60000. if self['queuingDurationMinutes'] == -1: raise ValueError('could not find queuingDurationMillis in: %s', json) if self['totalDurationMinutes'] == -1: raise ValueError('could not find totalDurationMillis in: %s', json) # Can be 'builds' (last 50) or 'allBuilds'. builds_key = 'allBuilds' builds = [] job_names = ['beam_PreCommit_Java_Cron', 'beam_PreCommit_Python_Cron', 'beam_PreCommit_Go_Cron'] for job_name in job_names: url = 'https://ci-beam.apache.org/job/%s/api/json' % job_name params = { 'tree': '%s[result,number,timestamp,actions[queuingDurationMillis,totalDurationMillis]]' % builds_key} r = requests.get(url, params=params) data = r.json() builds.extend([Build(job_name, build_json) for build_json in data[builds_key]]) df = pd.DataFrame(builds)

这段代码蕴含了三个值得展开的实现细节：

1. 时间戳换算：Jenkins 返回的timestamp是毫秒级 Unix 时间戳，代码先除以 1000 再交给pd.Timestamp.utcfromtimestamp，得到 UTC 时间用于后续时间窗过滤；
2. 排队/总耗时的来源：queuingDurationMillis与totalDurationMillis并不在构建 JSON 顶层，而是藏在actions数组中_class == 'jenkins.metrics.impl.TimeInQueueAction'的条目里。解析时以-1作为哨兵值，若构建缺失该 action 则直接抛出ValueError，避免脏数据进入统计；
3. 构建范围开关：builds_key注释明确说明可选'builds'（最近 50 次）或'allBuilds'，默认取'allBuilds'以最大化样本量。

采集对象是三个定时运行的 PreCommit 任务：beam_PreCommit_Java_Cron、beam_PreCommit_Python_Cron、beam_PreCommit_Go_Cron，分别对应 Beam 三大语言 SDK 的 PreCommit 测试。请求通过tree参数精确限定需要的字段（result,number,timestamp,actions[queuingDurationMillis,totalDurationMillis]），既满足 ASF Jenkins API 的强制要求，也大幅压缩了响应体积。

六、时间窗过滤：4 周 / 1 周 / 1 天三档切片

单元 3 基于"当前时刻"动态计算三个分析窗口，无需手工指定日期：

timestamp_cutoff = pd.Timestamp.utcnow().tz_convert(None) - pd.Timedelta(weeks=4) df_4weeks = df[df.timestamp >= timestamp_cutoff] timestamp_cutoff = pd.Timestamp.utcnow().tz_convert(None) - pd.Timedelta(weeks=1) df_1week = df[df.timestamp >= timestamp_cutoff] timestamp_cutoff = pd.Timestamp.utcnow().tz_convert(None) - pd.Timedelta(days=1) df_1day = df[df.timestamp >= timestamp_cutoff]

pd.Timestamp.utcnow()取当前 UTC 时间，tz_convert(None)去掉时区信息以与Build中无时区的 timestamp 对齐比较，随后依次回退 4 周、1 周、1 天生成截止点，用布尔掩码过滤出三个 DataFrame。三个窗口的划分与后续分位数统计一一对应，用于回答"近期（1 天/1 周）与中长期（4 周）的耗时是否恶化"。

七、耗时可视化：按任务绘制时间序列

单元 4 为每个任务单独画一张"时间-耗时"曲线，横轴为构建时间戳，纵轴同时绘制排队时长与总时长两条线：

# Graphs of precommit job times. for job_name in job_names: duration_df = df_4weeks[df_4weeks.job_name == job_name] duration_df = duration_df[['timestamp', 'queuingDurationMinutes', 'totalDurationMinutes']] ax = duration_df.plot(x='timestamp') ax.set_title(job_name)

duration_df.plot(x='timestamp')在 pandas 内部即调用 matplotlib 绘图，ax.set_title(job_name)以任务名作为图标题。由于queuingDurationMinutes与totalDurationMinutes的取值是Build解析时从毫秒换算出的分钟数，图上直接以"分钟"为纵轴单位，便于人工判读排队瓶颈与总耗时的变化趋势。

八、分位数统计：95 百分位画像

单元 5 是 Notebook 的核心分析输出，针对三个时间窗 × 三个任务，分别计算全部构建与仅 SUCCESS 构建的总耗时 95 百分位，以及排队耗时的 95 百分位：

# Get 95th percentile of precommit run times. test_dfs = {'4 weeks': df_4weeks, '1 week': df_1week, '1 day': df_1day} metrics = [] for sample_time, test_df in test_dfs.items(): for job_name in job_names: df_times = test_df[test_df.job_name == job_name] for percentile in [95]: total_all = np.percentile(df_times.totalDurationMinutes, q=percentile) total_success = np.percentile(df_times[df_times.result == 'SUCCESS'].totalDurationMinutes, q=percentile) queue = np.percentile(df_times.queuingDurationMinutes, q=percentile) metrics.append({'job_name': '%s %s %dth' % ( job_name.replace('beam_PreCommit_','').replace('_GradleBuild',''), sample_time, percentile), 'totalDurationMinutes_all': total_all, 'totalDurationMinutes_success_only': total_success, 'queuingDurationMinutes': queue, }) pd.DataFrame(metrics).sort_values('job_name')

几个值得注意的设计：

• 用result == 'SUCCESS'过滤出成功构建再计算分位数，从而把失败/中断构建对总耗时分布的扰动分离出来，totalDurationMinutes_all与totalDurationMinutes_success_only两列形成对照；
• 任务名在展示时被清洗：beam_PreCommit_前缀与_GradleBuild后缀被剥掉，只保留Java、Python、Go与时间窗、百分位组合（如'Java 4 weeks 95th'）；
• 结果通过pd.DataFrame(metrics).sort_values('job_name')输出为便于直接阅读的统计表。

为什么选 95 百分位而非平均值？对于 CI 耗时观测，平均值易被个别极端慢构建拉高，而 95 百分位更贴近"用户在绝大多数情况下会遭遇的等待时间"，是判断 PreCommit 任务健康度的稳健指标。

九、用例级分析：抓取单个测试的耗时与状态

如果说前五个单元回答"任务整体多慢"，单元 6 则下沉到"哪个测试用例最慢"。它通过 Jenkins 的testReportAPI 抓取每个构建的详细测试结果：

# Fetch individual test data (precommit) from Jenkins. MAX_FETCH_PER_JOB_TYPE = 5 test_results_raw = [] for job_name in list(df.job_name.unique()): if job_name == 'beam_PreCommit_Go_Cron': # TODO: Go builds are missing testReport data on Jenkins. continue build_nums = list(df.number[df.job_name == job_name].unique()) num_fetched = 0 for build_num in build_nums: url = 'https://ci-beam.apache.org/job/%s/%s/testReport/api/json?depth=1' % (job_name, build_num) print('.', end='') r = requests.get(url) if not r.ok: # Typically a 404 means that the job is still running. print('skipping (%s): %s' % (r.status_code, url)) continue raw_result = r.json() raw_result['job_name'] = job_name raw_result['build_num'] = build_num test_results_raw.append(raw_result) num_fetched += 1 if num_fetched >= MAX_FETCH_PER_JOB_TYPE: break print(' done')

这里有三处工程细节值得留意：

1. Go 任务被显式跳过：代码注释指出 Go 构建在 Jenkins 上缺少testReport数据（"TODO: Go builds are missing testReport data on Jenkins"），因此只采集 Java 与 Python；
2. 深度参数?depth=1：这是访问 ASF Jenkins API 必须携带的参数之一，同时保证suites与cases嵌套结构能被完整返回；
3. 404 的语义处理：注释说明"typically a 404 means that the job is still running"，即正在运行中的构建还没有生成测试报告，代码选择跳过而非报错，并用MAX_FETCH_PER_JOB_TYPE = 5限制每个任务最多抓取 5 个构建，控制请求总量、避免触发封禁。

十、用例数据规整与交互式 Top-N 分析

抓回的原始 JSON 嵌套在suites -> cases两层结构中，单元 6 的下半部分先定义TestResult类做扁平化，再完成聚合、排序与交互过滤：

# Analyze individual test results. class TestResult(dict): def __init__(self, job_name, build_num, json): self['job_name'] = job_name self['build_num'] = build_num self['name'] = json['name'] self['duration'] = json['duration'] self['className'] = json['className'] self['status'] = json['status'] test_results = [] for test_result_raw in test_results_raw: job_name = test_result_raw['job_name'] build_num = test_result_raw['build_num'] for suite in test_result_raw['suites']: for case in suite['cases']: test_results.append(TestResult(job_name, build_num, case)) df_tests = pd.DataFrame(test_results) df_tests = df_tests.drop(columns=['build_num']) df_tests = df_tests.groupby(['className', 'job_name', 'name', 'status'], as_index=False).max() df_tests = df_tests.sort_values('duration', ascending=False) def filter_test_results(job_name, status): res = df_tests if job_name != 'all': res = res[res.job_name == job_name] if status != 'all': res = res[res.status == status] return res.head(n=20) from ipywidgets import interact interact(filter_test_results, job_name=['all'] + list(df_tests.job_name.unique()), status=['all'] + list(df_tests.status.unique()))

分析逻辑可以拆解为四步：

1. 扁平化：遍历suites下的每个cases，把name（用例名）、className（所属类）、duration、status连同任务名抽出来，丢掉冗余的build_num；
2. 去重取极值：以(className, job_name, name, status)为键做groupby(...).max()，同一个测试在多个构建中出现时只保留最大耗时记录，聚焦"最坏情况"；
3. 排序：按duration降序排列，让最慢的测试排在最前；
4. 交互过滤：借助ipywidgets.interact生成任务名与状态两个下拉控件，回调filter_test_results支持job_name、status两个维度过滤（含all全选），并固定返回 Top 20。

运行后即可在 Notebook 内交互查看"Java/Python 任务中最慢的 20 个测试用例"，这是定位具体性能瓶颈的最后一环，也是排查 PreCommit 超时的直接入口。

十一、提交规范：清空 Cell 输出再提交

README 对向该目录提交改动提出了明确要求，属于仓库的协作约定：

To minimize file size, diffs, and ease reviews, please clear all cell output (cell -> all output -> clear) before committing.

即为了控制文件体积、缩小 diff 并方便评审，提交前必须执行 Jupyter 菜单中的Cell -> All Output -> Clear清空全部单元格输出。这一点对.ipynb尤其重要：Notebook 是 JSON 格式，输出内容（尤其是图表与大型 DataFrame）会以 base64 文本形式内嵌，未清理的输出会让每次运行结果都写进 diff，污染代码评审。仓库中 precommit_job_times.ipynb 所有单元格的outputs均为空数组，正是这一约定的实际体现。

十二、关联设施：Beam 测试指标体系的完整拼图

该 Notebook 并非孤立存在，它与仓库内其他测试基建共同构成指标闭环，读者可按需延伸：

• .test-infra/metrics/sync/jenkins/syncjenkins.py：以定时任务方式把 Jenkins 构建记录（含timing_queuingDurationMillis、timing_totalDurationMillis）写入 PostgreSQL 的jenkins_builds表，其配套的 README 提供了基于 Docker 的本地运行方式（docker run ... -e "JENSYNC_PORT=5432" ... syncjenkins.py），可视为 Notebook 的"长期归档版"；
• .test-infra/junitxml_report.py：解析 JUnitXML 格式测试报告，输出类名.用例名 状态文本流，适合离线对比 nosetests 与 pytest 的测试收集差异，与 Notebook 的在线 API 采集形成互补；
• .test-infra/metrics/grafana：包含大量 Grafana Dashboard 定义与 InfluxDB/PostgreSQL 存储配置，是这些测试指标在监控面板上的最终呈现层；
• CI.md：描述 Beam 整体 CI 环境（GitHub Actions 与 Jenkins 迁移历史），为理解 PreCommit 任务在整个发布与测试流程中的位置提供背景。

结语

.test-infra/jupyter 目录虽小，却浓缩了 Apache Beam 测试指标观测的完整方法论：从遵守 ASF Jenkins API 的tree/depth约束安全取数，到用TimeInQueueAction拆分排队与执行耗时，再到以 95 百分位和 Top-N 用例排序定位瓶颈。无论是想复现 Beam 的 CI 耗时分析，还是为自己的开源项目搭建类似的 Jenkins 指标观测 Notebook，本文梳理的代码路径、参数含义与提交规范都能直接照搬使用。

• 大数据
• 批处理
• 流处理
• 数据工程