钉钉考勤数据自动化同步：从API调用到数据落地的企业级实践

1. 项目概述：从零构建企业考勤数据自动化枢纽

最近在做一个企业内部的数据中台项目，其中一个核心需求就是把钉钉上的组织架构、人员信息和考勤假期数据给“搬”下来，进行二次分析和报表呈现。听起来好像就是调几个API的事儿？但真上手了才发现，这里面的门道不少。比如，如何高效、稳定地同步几十万人的组织树？考勤报表里的“假期”数据到底对应哪些复杂的假期类型？API的调用频率限制怎么破？这些都不是官方文档一句话能说清的。

这个项目的核心目标，就是建立一个自动化的数据管道，定时从钉钉拉取部门列表、人员列表以及智能考勤报表中的假期数据，并转化为结构清晰、可供业务系统（如BI报表、HR系统、财务系统）直接消费的数据源。它解决的是企业数据孤岛的问题——考勤数据不再封闭在钉钉里，而是能和企业自有的ERP、OA系统打通，为人力分析、成本核算、运营效率评估提供数据基础。

无论你是负责内部系统开发的工程师，还是需要处理大量人事数据的分析师，这个方案都能给你提供一个从接口鉴权、数据抓取、到清洗落地的完整参考。我会把过程中趟过的坑、优化的技巧，以及那些官方文档语焉不详的细节，都揉碎了讲清楚。

2. 整体架构设计与核心思路拆解

在动手写代码之前，我们先得把整个数据流的逻辑盘清楚。钉钉开放平台提供了丰富的接口，但直接盲目调用，很容易陷入混乱，后期维护更是噩梦。

2.1 为什么选择“部门-人员-考勤”这个数据链路？

这其实对应了企业数据管理的三个层次：组织（Department）、人（User）、事（Attendance）。部门列表构成了企业的骨架，人员列表填充了血肉，而考勤假期数据则反映了组织的“脉搏”和个体的“行为”。只同步人员，你不知道他属于哪个团队；只同步考勤，你无法进行部门维度的聚合分析。因此，这三者必须作为一个整体来考虑，并且存在明确的依赖关系：通常需要先获取部门ID，才能获取该部门下的人员；有了人员的userId，才能去查询其具体的考勤假期记录。

2.2 技术方案选型：脚本化还是平台化？

根据数据量和使用频率，主要有两种思路：

1. 轻量级脚本方案：适用于数据量不大（如数千人）、同步频率不高（如每日一次）的场景。可以用Python/Node.js写一个定时脚本，结合crontab或Windows任务计划程序运行。优点是开发部署快，依赖简单。
2. 平台化任务调度方案：适用于中大型企业，需要高可靠性、监控告警、失败重试、依赖管理等。可以选用Apache Airflow、DolphinScheduler，或者云厂商的数据开发平台（如阿里云DataWorks）。这套方案更健壮，但架构复杂。

本次分享，我会以Python脚本方案为主线，因为它最直观，理解了核心逻辑后，迁移到任何平台都只是“搬家”的问题。我们会用到requests库处理HTTP请求，pandas做数据转换，SQLAlchemy或直接SQL写入数据库。

2.3 核心挑战与应对策略

在设计之初，就要预见并规避以下几个关键问题：

• API限流：钉钉开放平台对大部分接口都有调用频率限制（例如，获取部门列表接口，每个企业每秒钟最多20次）。粗暴的循环调用会迅速触发限流。解决方案是加入适当的延迟（sleep），并对大批量数据（如遍历全公司人员）采用分页或异步请求。
• 数据一致性：如何识别新增、离职、调岗的人员？单纯的全量覆盖式同步，无法记录人员变动历史。一个常见的做法是，每次同步时，为每条数据打上“数据拉取时间戳”，并通过对比上一次的数据快照，来识别变更。
• 假期数据解析：智能考勤报表返回的假期数据是一个复杂的嵌套结构，包含了调休、年假、事假、婚假等多种类型，且格式可能随钉钉版本更新而变化。需要设计一个弹性、可扩展的解析器，而不是写死解析逻辑。
• 鉴权Token管理：调用钉钉API需要AccessToken，而这个Token有过期时间（通常2小时）。我们的同步程序必须能自动、安全地刷新Token，避免因Token失效导致同步任务失败。

3. 关键步骤一：钉钉开放平台准备与鉴权

万事开头难，第一步是获得进入钉钉数据的“门票”。

3.1 创建应用与获取凭证

首先，你需要有一个钉钉企业的管理员权限，或者请管理员协助操作。

1. 登录钉钉开放平台：访问钉钉开放平台官网，使用企业管理员账号登录。
2. 创建应用：在“应用开发” -> “企业内部开发”中，选择“创建应用”。应用类型选择“H5微应用”或“小程序”均可，因为我们只需要后端调用API的能力。填写应用名称、描述等信息。
3. 获取关键凭证：应用创建成功后，在应用详情页，你需要记录下以下三个核心信息：
   • AppKey与AppSecret：这是应用的身份证和密码，用于获取AccessToken。务必妥善保管AppSecret，不要泄露在客户端代码中。
   • AgentId：应用代理ID，在某些接口中可能会用到。
   • CorpId：企业ID，是你的企业在钉钉的唯一标识。

注意：确保在应用权限管理中，为你的应用添加必要的接口调用权限。对于本项目，至少需要：通讯录权限（读取部门、成员信息）和考勤权限（读取考勤数据）。提交发布后，管理员需审批通过，权限才会生效。

3.2 实现AccessToken的获取与缓存机制

AccessToken是调用几乎所有钉钉API的通行证。我们不能每次调用API都去获取一次，那样效率低下且容易触发限流。正确的做法是获取一次，缓存起来，在失效前重复使用。

import requests import time import json class DingTalkClient: def __init__(self, app_key, app_secret): self.app_key = app_key self.app_secret = app_secret self.access_token = None self.token_expire_time = 0 self.base_url = "https://oapi.dingtalk.com" def get_access_token(self): """获取并缓存AccessToken""" # 如果当前token存在且未过期，直接返回 if self.access_token and time.time() < self.token_expire_time: return self.access_token # 否则，重新获取 url = f"{self.base_url}/gettoken" params = { "appkey": self.app_key, "appsecret": self.app_secret } resp = requests.get(url, params=params) result = resp.json() if result.get("errcode") == 0: self.access_token = result["access_token"] # 钉钉返回的expires_in通常是7200秒（2小时），我们预留5分钟缓冲期 self.token_expire_time = time.time() + result["expires_in"] - 300 print(f"Token获取成功，有效期至{time.ctime(self.token_expire_time)}") return self.access_token else: raise Exception(f"获取AccessToken失败: {result}") # 使用示例 client = DingTalkClient(app_key="你的AppKey", app_secret="你的AppSecret") token = client.get_access_token()

实操心得：在实际生产环境中，建议将Token缓存到Redis或数据库中，尤其是在分布式多节点运行同步任务时，可以避免多个节点重复获取Token，也能保证Token的一致性。上面的内存缓存方式仅适用于单机脚本。

4. 关键步骤二：递归获取完整的部门树列表

钉钉的组织架构可能非常深，并且存在多个根部门。获取部门列表的关键在于理解其“递归”特性。

4.1 调用部门列表接口

钉钉提供了/department/list接口，它有一个关键参数fetch_child。当你传入一个部门ID时，设置fetch_child=True，可以获取其下的所有子部门。

def get_department_list(self, dept_id=1, fetch_child=True): """获取部门列表 :param dept_id: 父部门id，1代表根部门（公司） :param fetch_child: 是否递归获取子部门 """ url = f"{self.base_url}/department/list" token = self.get_access_token() params = { "access_token": token, "dept_id": dept_id, "fetch_child": fetch_child } resp = requests.get(url, params=params) result = resp.json() if result.get("errcode") == 0: departments = result.get("department", []) print(f"获取到部门数量: {len(departments)}") return departments else: print(f"获取部门列表失败: {result}") return [] # 获取整个公司的部门树 all_depts = client.get_department_list(dept_id=1, fetch_child=True)

4.2 处理部门数据并构建关系

接口返回的部门列表是一个扁平数组，每个部门对象包含id,name,parentid,order等字段。为了便于使用，我们通常需要将其转换为两种形式：

1. 父子关系列表：直接存入数据库，每行记录包含自己的ID和父部门ID，适合用SQL递归查询。
2. 树形结构：在内存中构建一棵树，用于前端展示或快速查找。这需要一次递归处理。

def build_dept_tree(self, dept_list, parent_id=1): """将扁平的部门列表构建成树形结构""" tree = [] for dept in dept_list: if dept['parentid'] == parent_id: node = { 'id': dept['id'], 'name': dept['name'], 'children': self.build_dept_tree(dept_list, dept['id']) } tree.append(node) return tree # 构建树形结构 dept_tree = build_dept_tree(all_depts) # 可以将其转换为JSON保存或使用

注意事项：

• 根部门ID：通常为1，但有些企业可能不同。如果不确定，可以不传dept_id参数，接口会返回所有部门，你可以通过parentid=0或parentid=1来找到根部门。
• 部门排序：返回数据中的order字段代表了在钉钉客户端里的显示顺序，同步时建议保留这个字段，以便在自建系统中还原相同的组织树视图。
• 接口限流：这个接口频率限制相对宽松，但如果你企业部门数量极多（上万），一次性拉取大量数据也要注意。通常一次调用就能获取全量。

5. 关键步骤三：分页获取全量人员列表

人员数据量往往比部门大得多，必须使用分页接口。这里有一个关键点：钉钉提供了两种获取人员列表的方式，一种是通过部门ID获取该部门下的人员（支持递归包含子部门），另一种是获取企业通讯录全员。我们通常选择后者，因为更直接。

5.1 使用获取企业全员接口

接口/user/list可以分页获取企业内所有员工。我们需要循环调用，直到获取所有页的数据。

def get_all_users(self, cursor=0, size=100): """分页获取企业全员列表 :param cursor: 游标，第一次传0 :param size: 分页大小，最大100 """ url = f"{self.base_url}/topapi/user/listsimple" token = self.get_access_token() all_users = [] has_more = True while has_more: payload = { "cursor": cursor, "size": size } headers = {'Content-Type': 'application/json'} # 注意这个接口是POST请求，参数在body中 resp = requests.post( f"{url}?access_token={token}", json=payload, headers=headers ) result = resp.json() if result.get("errcode") == 0: page_data = result.get("result", {}) user_list = page_data.get("list", []) all_users.extend(user_list) has_more = page_data.get("has_more", False) cursor = page_data.get("next_cursor", 0) print(f"已获取{len(all_users)}条人员记录...") # 避免请求过快，轻微延迟 time.sleep(0.1) else: print(f"获取人员列表失败: {result}") break return all_users # 获取所有人员 all_users_simple = client.get_all_users()

5.2 获取人员的详细信息

上面的listsimple接口返回的是简略信息（userId, name）。如果需要更详细的信息，如职位、邮箱、手机号（需有对应权限）、部门列表等，需要再调用/user/get接口，传入每个员工的userid。

def get_user_detail(self, userid): """获取单个成员的详细信息""" url = f"{self.base_url}/user/get" token = self.get_access_token() params = { "access_token": token, "userid": userid } resp = requests.get(url, params=params) return resp.json() # 批量获取详情（注意控制频率） detailed_users = [] for user in all_users_simple[:10]: # 示例：只取前10个获取详情 detail = client.get_user_detail(user['userid']) if detail.get('errcode') == 0: detailed_users.append(detail) time.sleep(0.05) # 重要：必须加延迟，否则极易触发限流！

核心避坑点：

• 速率限制：/user/get接口的频率限制非常严格（企业每秒钟最多20次）。直接循环调用上千次必然失败。必须在每次请求后加入time.sleep，建议至少0.05秒（即每秒不超过20次）。对于上万人员的企业，获取详情会非常慢，需要考虑异步IO或任务队列。
• 字段权限：手机号、邮箱等敏感字段，即使接口支持返回，也需要你的应用拥有“通讯录个人信息读权限”并获得员工授权。否则返回为空。
• 离职人员：默认接口可能不包含离职人员。如果需要，在获取详情时，可以尝试传入参数active=false（取决于具体接口版本），或使用专门的查询离职员工接口。

6. 关键步骤四：解析智能考勤报表中的假期数据

这是本项目最复杂的一环。智能考勤报表接口返回的数据结构复杂，且假期信息嵌套在深处。

6.1 理解考勤报表接口

我们主要使用/attendance/list接口。它需要传入一个复杂的请求体，指定查询的用户、时间范围等。

def get_attendance_report(self, userids, work_date_from, work_date_to): """获取考勤报表数据 :param userids: 员工id列表，每次最多50个 :param work_date_from: 查询起始工作日，格式 yyyy-MM-dd :param work_date_to: 查询结束工作日，格式 yyyy-MM-dd """ url = f"{self.base_url}/attendance/list" token = self.get_access_token() payload = { "workDateFrom": work_date_from, "workDateTo": work_date_to, "userIdList": userids, # 注意这里是数组 "offset": 0, "limit": 50 # 每页最大50 } headers = {'Content-Type': 'application/json'} resp = requests.post( f"{url}?access_token={token}", json=payload, headers=headers ) result = resp.json() if result.get("errcode") == 0: return result.get("recordresult", []) else: print(f"获取考勤报表失败: {result}") return []

6.2 深度解析假期字段

接口返回的每条考勤记录recordresult中，有一个字段叫做cal_accept_status（计算状态），但假期详情藏在另一个字段里，通常是cal_accept_ext或approve_list这样的扩展字段中，其结构是一个JSON字符串，需要再次解析。

假设我们从cal_accept_ext中解析出了假期数据leave_data，它可能长这样：

{ "leave": [ { "leave_name": "年假", "leave_code": "annual_leave", "leave_time": "2023-10-01", "leave_duration": 1.0, "leave_unit": "天", "leave_status": "已通过" }, { "leave_name": "调休", "leave_code": "compensatory_leave", "leave_time": "2023-10-01 上午", "leave_duration": 0.5, "leave_unit": "天", "leave_status": "已通过" } ] }

我们的解析代码需要足够健壮：

import json def parse_leave_data(attendance_record): """从单条考勤记录中解析假期数据""" leave_records = [] # 1. 尝试从主要字段获取 base_info = attendance_record.get('baseCheckTime', {}) user_id = attendance_record.get('userId') work_date = attendance_record.get('workDate') # 2. 关键：解析扩展字段中的假期信息 ext_info_str = attendance_record.get('cal_accept_ext') if ext_info_str: try: ext_info = json.loads(ext_info_str) # 假期数据可能在不同的键下，需要根据实际情况调整 leave_list = ext_info.get('leave', []) or ext_info.get('approveList', []) for leave_item in leave_list: record = { 'user_id': user_id, 'work_date': work_date, 'check_time': base_info.get('onTime'), # 上班时间 'leave_name': leave_item.get('leave_name'), 'leave_code': leave_item.get('leave_code'), 'leave_duration': float(leave_item.get('leave_duration', 0)), 'leave_unit': leave_item.get('leave_unit'), 'leave_status': leave_item.get('leave_status'), 'source_data': json.dumps(leave_item) # 保留原始数据，以防字段变更 } # 处理半天假、小时假的情况 if record['leave_unit'] == '半天' and record['leave_duration'] == 1: record['leave_duration_in_days'] = 0.5 elif record['leave_unit'] == '小时': record['leave_duration_in_days'] = record['leave_duration'] / 8.0 # 假设8小时工作制 else: record['leave_duration_in_days'] = record['leave_duration'] leave_records.append(record) except json.JSONDecodeError as e: print(f"解析扩展字段JSON失败: {e}, 原始数据: {ext_info_str}") return leave_records

经验之谈：

• 字段不固定：钉钉考勤报表的字段结构可能随版本更新或企业定制而变化。最稳妥的做法是，在解析逻辑中加入大量的日志，将原始响应和解析后的结构保存下来，定期审查。一旦发现解析失败，能快速定位是哪个字段发生了变化。
• 假期类型映射：不同企业对假期类型的命名可能不同（如“年假” vs “Annual Leave”）。建议在本地维护一个假期类型映射表，将钉钉返回的leave_code或leave_name映射到你系统内部统一的假期编码上，这样后续统计才不会出错。
• 分页与分批：考勤报表接口也支持分页（offset和limit），并且单次查询的用户数不能超过50。对于大规模查询，需要按用户分组、按时间分片，多批次调用。

7. 数据落地、调度与错误处理

把数据从接口里拿出来只是第一步，如何稳定、可靠地存下去，并形成一个自动化的流程，才是工程化的体现。

7.1 数据存储设计

建议至少创建三张核心表：

1. 部门表 (dim_dingtalk_dept)：存储部门ID、名称、父部门ID、排序、同步时间等。
2. 人员表 (dim_dingtalk_user)：存储用户ID、姓名、部门ID列表（可能是数组或JSON字符串）、职位、状态、同步时间等。
3. 考勤假期事实表 (fact_attendance_leave)：存储每次同步的假期流水，包含用户ID、日期、假期类型、时长、数据来源批次等。这是一张流水表，记录所有历史。

-- 以PostgreSQL为例的简化版表结构 CREATE TABLE dim_dingtalk_dept ( dept_id BIGINT PRIMARY KEY, name VARCHAR(255), parent_id BIGINT, dept_order BIGINT, sync_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE dim_dingtalk_user ( user_id VARCHAR(100) PRIMARY KEY, name VARCHAR(255), dept_ids JSONB, -- 存储所属部门ID列表 job_number VARCHAR(100), sync_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE fact_attendance_leave ( id SERIAL PRIMARY KEY, user_id VARCHAR(100), work_date DATE, leave_name VARCHAR(100), leave_code VARCHAR(50), leave_duration_in_days DECIMAL(5,2), sync_batch_id VARCHAR(50), -- 用于标识本次同步批次 sync_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP );

7.2 全流程脚本组装与定时调度

将上述所有步骤串联起来，形成一个完整的Python脚本。脚本的主要流程如下：

def main_sync_job(): """主同步任务函数""" batch_id = f"sync_{int(time.time())}" # 生成一个同步批次ID # 1. 初始化客户端，获取Token client = DingTalkClient(APP_KEY, APP_SECRET) # 2. 同步部门数据 print("开始同步部门数据...") departments = client.get_department_list() save_depts_to_db(departments, batch_id) # 3. 同步人员基础列表 print("开始同步人员基础列表...") users_simple = client.get_all_users() save_users_simple_to_db(users_simple, batch_id) # 4. (可选)分批获取人员详情，注意速率限制 print("开始同步人员详情...") detailed_users = [] for i in range(0, len(users_simple), 10): # 每10个一批，控制并发 batch = users_simple[i:i+10] for user in batch: detail = client.get_user_detail(user['userid']) if detail.get('errcode') == 0: detailed_users.append(detail) time.sleep(0.06) # 严格控制频率 save_users_detail_to_db(detailed_users, batch_id) # 5. 同步考勤假期数据（示例：同步昨天的数据） print("开始同步考勤假期数据...") yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d') # 假设我们只同步活跃用户 active_userids = [u['userid'] for u in users_simple] # 分批查询，每批最多50人 all_leave_records = [] for i in range(0, len(active_userids), 50): batch_ids = active_userids[i:i+50] attendance_data = client.get_attendance_report(batch_ids, yesterday, yesterday) for record in attendance_data: leave_records = parse_leave_data(record) all_leave_records.extend(leave_records) time.sleep(1) # 考勤接口也需适当延迟 save_leave_to_db(all_leave_records, batch_id) print(f"同步完成！批次ID: {batch_id}") if __name__ == "__main__": main_sync_job()

然后，在Linux服务器上，使用crontab设置定时任务，例如每天凌晨2点执行一次：

# 编辑crontab crontab -e # 添加一行 0 2 * * * /usr/bin/python3 /path/to/your/dingtalk_sync.py >> /path/to/log/sync.log 2>&1

7.3 错误处理与监控

一个健壮的同步程序必须考虑失败情况。

• 网络异常与重试：使用try...except包裹网络请求，遇到连接超时、HTTP错误时，进行有限次数的重试（如3次）。
• Token失效重试：在调用业务API时，如果返回错误码40001（Token无效），应自动触发一次Token刷新，然后用新Token重试该请求。
• 数据写入去重：对于部门、人员这类维度表，采用“upsert”（存在则更新，不存在则插入）操作，避免重复记录。对于考勤流水表，可以使用sync_batch_id和唯一键（user_id,work_date,leave_code）来避免同一批次内重复插入。
• 日志与告警：将关键步骤、获取的数据量、遇到的错误都详细记录到日志文件中。对于同步失败、数据量异常（如今天获取到0条考勤记录）等情况，可以集成邮件、钉钉机器人Webhook发送告警通知。

8. 常见问题排查与性能优化技巧

在实际运行中，你肯定会遇到各种各样的问题。下面是我踩过坑后总结的一些排查思路和优化点。

8.1 高频问题速查表

问题现象	可能原因	排查步骤与解决方案
获取Token失败，返回40001	AppKey或AppSecret错误；应用未发布/未授权。	1. 检查开放平台应用凭证是否复制正确。 2. 登录钉钉管理后台，检查应用是否已“发布”且员工已“授权”。 3. 确认使用的CorpId、AppKey、AppSecret三者对应同一个应用。
调用部门/人员接口返回88错误码	接口调用频率超限。	1.立即在代码中增加请求间隔（sleep）。 2. 检查是否在短时间内在循环中无延迟地调用了/user/get这类敏感接口。 3. 考虑将大批量任务分散到更长的时间窗口内执行。
获取人员详情时，手机号/邮箱为空	应用权限不足或员工未授权。	1. 在钉钉开放平台，检查应用权限列表是否包含“手机号码信息”和“邮箱等个人信息”的读取权限。 2. 管理员需在管理后台完成应用授权，且员工个人可能需要在手机钉钉上确认授权。
考勤报表接口返回成功，但recordresult为空数组	查询时间范围内无考勤数据；用户不在考勤组中。	1. 确认workDateFrom和workDateTo格式为yyyy-MM-dd。 2. 确认查询的用户在指定日期是否已加入公司的考勤组。 3. 尝试查询一个已知有打卡记录的日期进行测试。
解析考勤扩展字段(cal_accept_ext)时JSON解码错误	字段内容可能不是标准JSON，或是空字符串，或结构已变更。	1. 在解析前先用if ext_info_str and ext_info_str.strip():判断非空。 2. 将原始字符串打印到日志中，人工检查其格式。 3. 编写更健壮的解析逻辑，使用try...except包裹，并记录解析失败的原始数据供后续分析。
同步脚本运行一段时间后内存占用过高	未及时清理中间变量；数据量过大一次性加载。	1. 对于大批量数据（如上万人员），不要一次性全放在列表里。采用流式处理，获取一批，处理一批，写入数据库一批，然后清空该批数据。 2. 使用数据库的批量插入（如executemany）而非逐条插入。

8.2 性能优化实战建议

1. 并发请求（谨慎使用）：对于获取人员详情这种高延迟操作，可以考虑使用concurrent.futures.ThreadPoolExecutor创建线程池进行有限并发（例如5-10个线程）。但必须注意，钉钉的限流是针对整个企业的，并发过高会立刻触发限流，导致所有请求失败。建议先以极低的并发度（如2-3）测试，观察返回的错误码，再逐步调整。
2. 增量同步：每天全量同步部门和人员可能浪费资源。可以记录每个部门或人员的更新时间（钉钉部分接口返回update_time），下次同步时只拉取更新时间大于上次同步时间的数据。考勤数据则天然按天增量同步。
3. 连接池与超时设置：使用requests.Session()来复用HTTP连接，提升效率。同时，务必设置合理的timeout参数（如连接超时5秒，读取超时30秒），避免脚本因网络问题无限挂起。
4. 数据库操作优化：使用ORM（如SQLAlchemy）的批量插入方法，或者直接构造批量插入的SQL语句（如INSERT INTO ... VALUES (...), (...), (...) ON CONFLICT ...），这比循环单条插入快一个数量级。

8.3 一个容易被忽略的细节：假期时长的标准化

钉钉返回的假期时长，单位可能是“天”、“半天”、“小时”。在内部统计时，我们需要一个标准单位（如“人天”）。

• “1天” = 1人天
• “1半天” = 0.5人天
• “X小时” = X / 标准日工作小时数（例如8）人天

在数据库设计时，可以同时存储原始值和换算后的标准值。例如：

leave_duration_original DECIMAL(5,2), -- 原始数值 leave_unit_original VARCHAR(10), -- 原始单位 leave_duration_standard DECIMAL(5,2) -- 换算为标准人天

这样既保留了原始数据，又方便了后续的聚合计算。