Pandas reset_index():重置 DataFrame 索引的完整指南
Updated on
在使用 pandas DataFrame 的过程中,经常会遇到“乱掉”的、非连续的索引。经历 groupby、行过滤或排序之后,索引可能出现断裂的缺口、重复值,或者变成没有意义的标签。这些不规则索引会让数据更难处理,破坏其他库对索引的默认假设,并在导出或可视化时产生令人困惑的输出。
reset_index() 方法就是为了解决这个问题:它会把 DataFrame 恢复为干净、连续的整数索引。无论你是在清洗数据流水线、为机器学习准备数据集,还是只是需要可预测的行号,理解如何正确重置索引都是高效 pandas 工作流的关键。
本指南将从基础语法讲到高级 MultiIndex 操作,帮助你彻底掌握 pandas 中的索引处理。
理解为什么需要 reset_index()
创建 DataFrame 时,pandas 会自动分配整数索引(0, 1, 2, ...)。但很多常见操作会破坏这种连续顺序:
过滤之后:使用布尔索引过滤行时,保留下来的行仍然保留原始索引值,于是出现空洞。
import pandas as pd
df = pd.DataFrame({
'name': ['Alice', 'Bob', 'Charlie', 'David'],
'score': [85, 92, 78, 95]
})
# Filter high scorers - index becomes [1, 3]
high_scorers = df[df['score'] > 80]
print(high_scorers)
# name score
# 1 Bob 92
# 3 David 95groupby 之后:分组会把列值变成索引层级,而你往往希望把它们再变回普通列。
# Group by category - category becomes the index
sales = pd.DataFrame({
'category': ['A', 'B', 'A', 'B'],
'revenue': [100, 150, 200, 175]
})
grouped = sales.groupby('category')['revenue'].sum()
print(grouped)
# category
# A 300
# B 325
# Name: revenue, dtype: int64排序之后:按值排序会重新排列行,但会保留原始索引。
sorted_df = df.sort_values('score')
print(sorted_df)
# name score
# 2 Charlie 78
# 0 Alice 85
# 1 Bob 92
# 3 David 95set_index() 之后:当你把某列提升为索引,之后又想把它还原为普通列时。
这些场景会产生:不从 0 开始的索引、非连续数字索引,或用类别值充当索引等问题。reset_index() 会把秩序恢复回来。
reset_index() 的基础语法
基础语法非常直接:
df.reset_index(drop=False, inplace=False, level=None, col_level=0, col_fill='')关键参数:
drop(bool):若为True,丢弃旧索引;若为False(默认),将旧索引转换为一列。inplace(bool):若为True,原地修改 DataFrame;若为False(默认),返回新的 DataFrame。level(int/str/list):用于 MultiIndex DataFrame,指定要重置的索引层级。
基础示例:
df = pd.DataFrame({
'value': [10, 20, 30]
}, index=[5, 10, 15])
# Reset to sequential index
df_reset = df.reset_index()
print(df_reset)
# index value
# 0 5 10
# 1 10 20
# 2 15 30默认情况下,旧索引会成为名为 "index" 的新列。
drop=True vs drop=False:何时保留或丢弃旧索引
drop 参数决定是否将旧索引保留为一列。
使用 drop=False(默认):当旧索引包含有意义的信息时:
# Time series data - preserve the date index
dates = pd.date_range('2024-01-01', periods=3)
df = pd.DataFrame({'sales': [100, 150, 200]}, index=dates)
df_reset = df.reset_index()
print(df_reset)
# index sales
# 0 2024-01-01 100
# 1 2024-01-02 150
# 2 2024-01-03 200
# Now you can filter by date as a column
recent = df_reset[df_reset['index'] >= '2024-01-02']使用 drop=True:当旧索引只是没有语义的行号时:
# After filtering - old index numbers are meaningless
filtered = df[df['sales'] > 120]
print(filtered)
# sales
# 2024-01-02 150
# 2024-01-03 200
# Drop old index, create fresh sequential one
clean = filtered.reset_index(drop=True)
print(clean)
# sales
# 0 150
# 1 200常见模式:在 groupby 聚合后,通常希望 drop=False,把分组列转换回普通列:
sales = pd.DataFrame({
'region': ['North', 'South', 'North', 'South'],
'product': ['A', 'A', 'B', 'B'],
'revenue': [100, 150, 200, 175]
})
# GroupBy makes region and product the index
summary = sales.groupby(['region', 'product'])['revenue'].sum()
print(summary)
# region product
# North A 100
# B 200
# South A 150
# B 175
# Name: revenue, dtype: int64
# Reset to get region and product back as columns
summary_df = summary.reset_index()
print(summary_df)
# region product revenue
# 0 North A 100
# 1 North B 200
# 2 South A 150
# 3 South B 175inplace 参数:直接修改 DataFrame
默认情况下,reset_index() 会返回一个新 DataFrame,而不会修改原对象。设置 inplace=True 则会直接在原对象上修改。
df = pd.DataFrame({'value': [1, 2, 3]}, index=[10, 20, 30])
# Default: returns new DataFrame
df_new = df.reset_index(drop=True)
print(df.index) # Still [10, 20, 30]
print(df_new.index) # RangeIndex(start=0, stop=3, step=1)
# inplace=True: modifies df directly
df.reset_index(drop=True, inplace=True)
print(df.index) # RangeIndex(start=0, stop=3, step=1)何时使用 inplace=True:
- 内存受限环境,拷贝开销很大
- 线性步骤操作,不需要保留原 DataFrame
何时使用 inplace=False(默认):
- 方法链(method chaining)风格
- 需要保留原数据
- 可读性与调试更好(显式赋值)
多数 pandas 开发者更偏好 inplace=False 以让代码更清晰:
# Clearer: explicit assignment
df = df.reset_index(drop=True)
# Less clear: invisible modification
df.reset_index(drop=True, inplace=True)使用 level 参数重置 MultiIndex DataFrame
MultiIndex(层级索引)需要更精细的处理。level 参数允许你选择性地重置某些索引层级。
创建 MultiIndex:
# MultiIndex from groupby
sales = pd.DataFrame({
'region': ['East', 'East', 'West', 'West'],
'quarter': ['Q1', 'Q2', 'Q1', 'Q2'],
'revenue': [100, 150, 200, 175]
})
multi_df = sales.set_index(['region', 'quarter'])
print(multi_df)
# revenue
# region quarter
# East Q1 100
# Q2 150
# West Q1 200
# Q2 175重置所有层级(默认行为):
# Reset both levels to columns
reset_all = multi_df.reset_index()
print(reset_all)
# region quarter revenue
# 0 East Q1 100
# 1 East Q2 150
# 2 West Q1 200
# 3 West Q2 175按位置重置指定层级:
# Reset only outer level (region)
reset_outer = multi_df.reset_index(level=0)
print(reset_outer)
# region revenue
# quarter
# Q1 East 100
# Q2 East 150
# Q1 West 200
# Q2 West 175按名称重置指定层级:
# Reset only quarter, keep region as index
reset_quarter = multi_df.reset_index(level='quarter')
print(reset_quarter)
# quarter revenue
# region
# East Q1 100
# East Q2 150
# West Q1 200
# West Q2 175重置多个指定层级:
# Create 3-level MultiIndex
df = pd.DataFrame({
'country': ['USA', 'USA', 'UK', 'UK'],
'state': ['CA', 'TX', 'London', 'Manchester'],
'city': ['LA', 'Austin', 'City', 'City'],
'population': [4000000, 950000, 9000000, 550000]
})
three_level = df.set_index(['country', 'state', 'city'])
# Reset only country and city, keep state
reset_some = three_level.reset_index(level=['country', 'city'])
print(reset_some)
# country city population
# state
# CA USA LA 4000000
# TX USA Austin 950000
# London UK City 9000000
# Manchester UK City 550000groupby 之后使用 reset_index()
reset_index() 最常见的用途之一,就是把 groupby 的结果变回常规 DataFrame。
简单聚合:
sales = pd.DataFrame({
'category': ['Electronics', 'Clothing', 'Electronics', 'Clothing'],
'revenue': [500, 300, 700, 400]
})
# groupby().sum() creates category as index
grouped = sales.groupby('category')['revenue'].sum()
print(grouped)
# category
# Clothing 700
# Electronics 1200
# Name: revenue, dtype: int64
# Convert to DataFrame with category as column
result = grouped.reset_index()
print(result)
# category revenue
# 0 Clothing 700
# 1 Electronics 1200使用 agg() 做多个聚合:
# Multiple aggregation functions
agg_result = sales.groupby('category')['revenue'].agg(['sum', 'mean', 'count'])
print(agg_result)
# sum mean count
# category
# Clothing 700 350 2
# Electronics 1200 600 2
# Reset to get category back as column
agg_df = agg_result.reset_index()
print(agg_df)
# category sum mean count
# 0 Clothing 700 350 2
# 1 Electronics 1200 600 2按多列分组:
sales = pd.DataFrame({
'region': ['North', 'South', 'North', 'South'],
'product': ['A', 'A', 'B', 'B'],
'units': [100, 150, 200, 175],
'revenue': [1000, 1500, 2000, 1750]
})
# Group by multiple columns
multi_group = sales.groupby(['region', 'product']).agg({
'units': 'sum',
'revenue': 'mean'
})
print(multi_group)
# units revenue
# region product
# North A 100 1000.0
# B 200 2000.0
# South A 150 1500.0
# B 175 1750.0
# Reset MultiIndex to columns
final = multi_group.reset_index()
print(final)
# region product units revenue
# 0 North A 100 1000.0
# 1 North B 200 2000.0
# 2 South A 150 1500.0
# 3 South B 175 1750.0方法链模式:
# Common pattern: groupby → agg → reset_index in one chain
summary = (sales
.groupby(['region', 'product'])
.agg({'revenue': 'sum', 'units': 'mean'})
.reset_index()
)过滤与切片之后使用 reset_index()
过滤与切片会保留原始索引,常常导致带缺口的非连续索引。
布尔过滤之后:
students = pd.DataFrame({
'name': ['Alice', 'Bob', 'Charlie', 'David', 'Eve'],
'score': [85, 92, 78, 95, 88],
'grade': ['B', 'A', 'C', 'A', 'B']
})
# Filter for A grades - index becomes [1, 3]
a_students = students[students['grade'] == 'A']
print(a_students)
# name score grade
# 1 Bob 92 A
# 3 David 95 A
# Reset for clean sequential index
a_students_clean = a_students.reset_index(drop=True)
print(a_students_clean)
# name score grade
# 0 Bob 92 A
# 1 David 95 Ailoc 切片之后:
# Take middle 3 rows
middle = students.iloc[1:4]
print(middle)
# name score grade
# 1 Bob 92 A
# 2 Charlie 78 C
# 3 David 95 A
# Reset index
middle_reset = middle.reset_index(drop=True)
print(middle_reset)
# name score grade
# 0 Bob 92 A
# 1 Charlie 78 C
# 2 David 95 A多个过滤条件之后:
# Complex filtering
high_performers = students[
(students['score'] > 85) &
(students['grade'].isin(['A', 'B']))
]
# Index is now [1, 3, 4] - not sequential
print(high_performers.index.tolist()) # [1, 3, 4]
# Clean reset
high_performers = high_performers.reset_index(drop=True)
print(high_performers.index.tolist()) # [0, 1, 2]reset_index() vs set_index():互补操作
reset_index() 与 set_index() 是互为逆操作:set_index() 把列提升为索引;reset_index() 把索引降级回列。
set_index() → reset_index() 往返:
df = pd.DataFrame({
'employee_id': [101, 102, 103],
'name': ['Alice', 'Bob', 'Charlie'],
'salary': [75000, 82000, 68000]
})
# Promote employee_id to index
indexed = df.set_index('employee_id')
print(indexed)
# name salary
# employee_id
# 101 Alice 75000
# 102 Bob 82000
# 103 Charlie 68000
# Restore employee_id as column
restored = indexed.reset_index()
print(restored)
# employee_id name salary
# 0 101 Alice 75000
# 1 102 Bob 82000
# 2 103 Charlie 68000何时使用哪个:
| Operation | Use Case |
|---|---|
set_index() | 通过键快速查找(.loc[key])、时间序列对齐、基于当前列做 groupby |
reset_index() | 导出 CSV/Excel、机器学习(算法常期望数值索引)、可视化、基于当前索引做 groupby |
实用工作流示例:
# Start with employee_id as regular column
employees = pd.DataFrame({
'employee_id': [101, 102, 103, 104],
'department': ['Sales', 'Sales', 'Engineering', 'Engineering'],
'salary': [75000, 82000, 95000, 88000]
})
# Set index for fast lookups
employees_indexed = employees.set_index('employee_id')
# Fast lookup by employee ID
bob_salary = employees_indexed.loc[102, 'salary'] # 82000
# Reset index to group by department
summary = (employees_indexed
.reset_index()
.groupby('department')['salary']
.mean()
.reset_index()
)
print(summary)
# department salary
# 0 Engineering 91500
# 1 Sales 78500重置带名称的索引
当 DataFrame 的索引带有名称(通过 index.name),reset_index() 会用该名称作为新列名。
命名索引示例:
# Create DataFrame with named index
df = pd.DataFrame({
'temperature': [72, 75, 68, 80]
}, index=pd.Index(['2024-01-01', '2024-01-02', '2024-01-03', '2024-01-04'], name='date'))
print(df)
# temperature
# date
# 2024-01-01 72
# 2024-01-02 75
# 2024-01-03 68
# 2024-01-04 80
# Reset - name 'date' becomes column name
reset_df = df.reset_index()
print(reset_df)
# date temperature
# 0 2024-01-01 72
# 1 2024-01-02 75
# 2 2024-01-03 68
# 3 2024-01-04 80带名称的 MultiIndex:
# Create MultiIndex with names
arrays = [
['A', 'A', 'B', 'B'],
['X', 'Y', 'X', 'Y']
]
index = pd.MultiIndex.from_arrays(arrays, names=['category', 'subcategory'])
df = pd.DataFrame({'value': [10, 20, 30, 40]}, index=index)
print(df)
# value
# category subcategory
# A X 10
# Y 20
# B X 30
# Y 40
# Reset - names become column names
reset_df = df.reset_index()
print(reset_df)
# category subcategory value
# 0 A X 10
# 1 A Y 20
# 2 B X 30
# 3 B Y 40在 reset 后重命名:
# Reset and immediately rename
reset_renamed = df.reset_index().rename(columns={'category': 'main_cat'})
print(reset_renamed)
# main_cat subcategory value
# 0 A X 10
# 1 A Y 20
# 2 B X 30
# 3 B Y 40常见模式与最佳实践
模式 1:Groupby 聚合流水线
# Standard pattern for groupby analysis
result = (df
.groupby(['category', 'region'])
.agg({'sales': 'sum', 'quantity': 'mean'})
.reset_index()
.sort_values('sales', ascending=False)
)模式 2:过滤后清洗索引
# Filter and reset in pipeline
clean_data = (df
[df['status'] == 'active']
.reset_index(drop=True)
)模式 3:保留时间序列索引信息
# Keep date index as column for plotting
plot_data = timeseries_df.reset_index()
plot_data.plot(x='date', y='value')模式 4:避免索引列名冲突
# If 'index' column already exists, reset_index creates 'level_0'
df = pd.DataFrame({'index': [1, 2, 3], 'value': [10, 20, 30]})
df_reset = df.reset_index()
print(df_reset.columns.tolist()) # ['level_0', 'index', 'value']
# Better: drop the old index if it's meaningless
df_reset = df.reset_index(drop=True)
print(df_reset.columns.tolist()) # ['index', 'value']模式 5:导出友好的 DataFrame
# Reset before saving to CSV to avoid extra index column
df.reset_index(drop=True).to_csv('output.csv', index=False)参数对比表
| Parameter | Default | Effect | When to Use |
|---|---|---|---|
drop=False | Yes | 将旧索引转换为列 | 索引包含有意义的数据(日期、ID、类别) |
drop=True | No | 丢弃旧索引并创建新的连续索引 | 旧索引只是行号,没有语义价值 |
inplace=False | Yes | 返回新 DataFrame,原对象不变 | 方法链、需要保留原数据 |
inplace=True | No | 原地修改并返回 None | 更省内存、线性步骤操作 |
level=None | Yes | 重置所有索引层级 | 单层索引或想重置整个 MultiIndex |
level=0 or level='name' | No | 重置指定索引层级 | MultiIndex 但想保留部分层级 |
col_level=0 | Yes | 指定 MultiIndex 列的列层级 | 高级:列本身也是 MultiIndex |
col_fill='' | Yes | 填充缺失的列名 | 高级:MultiIndex 列的边界情况 |
真实世界示例
示例 1:为机器学习准备数据
# Load dataset with messy index
import pandas as pd
from sklearn.model_selection import train_test_split
df = pd.read_csv('sales_data.csv')
# After filtering and feature engineering, index is fragmented
df_filtered = df[df['valid'] == True].copy()
df_filtered['revenue_per_unit'] = df_filtered['revenue'] / df_filtered['units']
# Reset index before train/test split
# Many ML libraries expect clean 0-indexed data
df_clean = df_filtered.reset_index(drop=True)
X = df_clean.drop('target', axis=1)
y = df_clean['target']
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)示例 2:时间序列重采样与分析
# Daily sales data
sales = pd.DataFrame({
'date': pd.date_range('2024-01-01', periods=365, freq='D'),
'revenue': range(365)
})
sales_ts = sales.set_index('date')
# Resample to monthly totals
monthly = sales_ts.resample('M')['revenue'].sum()
print(monthly.head())
# date
# 2024-01-31 465
# 2024-02-29 1305
# 2024-03-31 2170
# Name: revenue, dtype: int64
# Reset index to get date back as column for plotting
monthly_df = monthly.reset_index()
monthly_df.columns = ['month', 'total_revenue']
# Now easy to plot with libraries that expect column names
import matplotlib.pyplot as plt
monthly_df.plot(x='month', y='total_revenue', kind='bar')示例 3:用于报表的多层聚合
# Complex business report
transactions = pd.DataFrame({
'region': ['East', 'East', 'West', 'West', 'East', 'West'],
'product': ['A', 'B', 'A', 'B', 'A', 'B'],
'quarter': ['Q1', 'Q1', 'Q1', 'Q1', 'Q2', 'Q2'],
'revenue': [1000, 1500, 2000, 1750, 1200, 1800]
})
# Multi-level groupby
report = (transactions
.groupby(['region', 'quarter', 'product'])
.agg({
'revenue': ['sum', 'mean', 'count']
})
.reset_index()
)
# Flatten column MultiIndex
report.columns = ['_'.join(col).strip('_') for col in report.columns]
print(report)
# region quarter product revenue_sum revenue_mean revenue_count
# 0 East Q1 A 1000 1000.0 1
# 1 East Q1 B 1500 1500.0 1
# 2 East Q2 A 1200 1200.0 1
# 3 West Q1 A 2000 2000.0 1
# 4 West Q1 B 1750 1750.0 1
# 5 West Q2 B 1800 1800.0 1示例 4:使用 PyGWalker 做可视化
import pandas as pd
import pygwalker as pyg
# After complex data transformations
df = pd.read_csv('metrics.csv')
summary = (df
.groupby(['category', 'month'])
.agg({'value': 'mean', 'count': 'sum'})
.reset_index() # Critical: PyGWalker works better with flat DataFrames
)
# Create interactive visualization
# reset_index() ensures clean column structure for drag-and-drop interface
walker = pyg.walk(summary)PyGWalker (opens in a new tab) 是一个开源 Python 库,可以把你的 pandas DataFrame 变成交互式、类似 Tableau 风格的可视化。对 MultiIndex 聚合结果先用 reset_index() 扁平化之后,PyGWalker 能提供拖拽式界面,让你无需额外编写绘图代码就能探索数据。当你想快速可视化分组结果、或与非技术干系人分享交互式看板时,这会特别有用。
常见错误与避免方法
错误 1:groupby 之后忘记 reset
# Wrong: grouped result has category as index
grouped = df.groupby('category')['value'].sum()
# Trying to access category as column fails
grouped['category'] # KeyError!
# Correct: reset to convert index to column
grouped_df = df.groupby('category')['value'].sum().reset_index()
grouped_df['category'] # Works!错误 2:生成重复列名
# DataFrame already has 'index' column
df = pd.DataFrame({'index': [1, 2, 3], 'value': [10, 20, 30]})
# reset_index() creates 'level_0' to avoid collision
df_reset = df.reset_index()
print(df_reset.columns.tolist()) # ['level_0', 'index', 'value']
# Solution: use drop=True if old index is meaningless
df_reset = df.reset_index(drop=True)
print(df_reset.columns.tolist()) # ['index', 'value']错误 3:使用 inplace 却不了解它会返回 None
# Wrong: assigns None to df
df = df.reset_index(drop=True, inplace=True)
print(df) # None
# Correct: don't assign when using inplace
df.reset_index(drop=True, inplace=True)
# Or better: use default behavior
df = df.reset_index(drop=True)错误 4:导出前没有丢弃无意义索引
# Wrong: creates extra 'Unnamed: 0' column in CSV
df.to_csv('output.csv')
# Correct: reset and specify index=False
df.reset_index(drop=True).to_csv('output.csv', index=False)错误 5:MultiIndex 中重置了错误层级
# MultiIndex: [region, product]
multi_df = df.set_index(['region', 'product'])
# Wrong: resets inner level (product), keeps region
wrong = multi_df.reset_index(level=1)
# Correct: reset outer level (region) if that's what you want
correct = multi_df.reset_index(level=0)
# Or reset both
both = multi_df.reset_index()FAQ
reset_index() 在 pandas 里是做什么的?
reset_index() 会把 DataFrame 当前的索引转换回普通列,并创建新的默认整数索引(0, 1, 2, ...)。这在 groupby、过滤、排序等操作破坏连续索引之后尤其重要。默认情况下,它会把旧索引保留为新列;你也可以用 drop=True 直接丢弃旧索引。
什么时候应该用 reset_index(drop=True)?
当现有索引不包含任何有意义的信息,你只想要一个从 0 开始的干净、连续整数索引时,使用 reset_index(drop=True)。这常见于过滤行之后、按值排序之后,或者索引只是上一步操作遗留的行号。如果索引包含你需要保留的信息(如日期、ID、类别),则使用 drop=False(默认)把索引转成列。
如何在 pandas 中重置 MultiIndex?
对于 MultiIndex DataFrame,直接使用不带参数的 reset_index() 可以把所有索引层级都转换回列。若只想重置部分层级,使用 level 参数:df.reset_index(level=0) 重置最外层,或 df.reset_index(level='level_name') 按名称重置。也可以传入列表来重置多个层级:df.reset_index(level=[0, 2])。
reset_index() 和 set_index() 有什么区别?
reset_index() 与 set_index() 是互为逆操作。set_index() 会把一列或多列提升为索引,适用于快速查找与时间序列相关操作;reset_index() 则把索引降回普通列,并创建新的默认整数索引。需要索引驱动的操作时用 set_index();需要把索引值当列用于分组、导出或可视化时用 reset_index()。
为什么 reset_index() 后会出现 'level_0' 列?
当 DataFrame 已经存在名为 'index' 的列时,调用 reset_index() 会产生 'level_0'(以及 'level_1' 等)来避免覆盖现有列名。要避免这种情况,你可以在 reset 前重命名现有的 'index' 列;或者当你不需要保留旧索引时,使用 reset_index(drop=True)。
reset_index() 应该使用 inplace=True 吗?
多数情况下建议使用 inplace=False(默认),让代码更清晰、更易维护,例如:df = df.reset_index()。inplace=True 会直接修改原 DataFrame、避免拷贝,在大数据场景可能更省内存,但它会返回 None,也更不利于调试。现代 pandas 风格通常更推崇显式赋值而非 inplace 操作。
pandas 里 groupby 之后如何重置索引?
groupby 之后,被分组的列会成为索引。调用 .reset_index() 可以把它们转换回普通列:df.groupby('category')['value'].sum().reset_index()。这是让 groupby 结果便于继续分析、导出或可视化的标准模式。常见流水线是:df.groupby(cols).agg(functions).reset_index()。
结论
熟练掌握 reset_index() 对于高效进行 pandas 数据处理至关重要。无论你是在过滤后清理数据、把 groupby 结果转换回扁平 DataFrame,还是为机器学习与可视化准备数据集,清楚何时以及如何重置索引都能让你的工作流更顺畅。
关键要点:
- 旧索引无意义时用
drop=True;需要保留索引信息时用drop=False - groupby 之后,
reset_index()能把分组列转换回普通列 - 对 MultiIndex,使用
level参数选择性重置部分层级 - 为了代码更清晰,优先使用
inplace=False(默认)并显式赋值 - 导出 CSV 或传入可视化库之前,通常应先重置索引
将 reset_index() 与 groupby()、set_index()、过滤等操作结合,你就能构建干净、可维护的数据转换流水线,并稳定产出可直接分析的 DataFrame。