本文详细介绍了如何利用pandas和正则表达式处理非标准格式的csv文件，该文件数据以堆叠方式存储，并由空行分隔。通过分块读取、解析和横向合并，我们将实现将多列信号数据统一到单个dataframe中，其中时间戳作为主索引，每个信号作为独立列，极大地提高了数据可用性。
在数据分析和处理的日常工作中，我们经常会遇到各种非标准格式的数据源。其中一种常见但具有挑战性的情况是，数据以“堆叠”或“块状”的形式存储在单个文件中，不同数据块之间由特定的分隔符（例如空行）隔开，每个块内部又包含自己的元数据和实际数据。这种格式使得直接使用 pandas.read_csv 等常规函数难以一次性导入。本文将详细阐述如何利用Pandas库结合正则表达式，高效地将此类堆叠式CSV文件重塑为规范的DataFrame，以便于后续分析。
问题描述与原始数据格式
假设我们有一个CSV文件，其内容结构如下所示。每个信号的数据都作为一个独立的块堆叠在一起，并通过一个空行（实际上是 , 后面跟着换行符，因为数据是两列）进行分隔。每个数据块的头部包含信号的元信息，例如 Trace Name、Signal 等，随后是 Timestamp 和 Value 的实际数据。
Trace Name,SignalName1 Signal,Timestamp,Value 2023-10-04 15:36:43.757193 EDT,13 2023-10-04 15:36:43.829083 EDT,14 2023-10-04 15:36:43.895651 EDT,17 2023-10-04 15:36:43.931145 EDT,11 , Trace Name,SignalName2 Signal, Timestamp,Value 2023-10-04 15:36:43.757193 EDT,131 2023-10-04 15:36:43.829083 EDT,238 2023-10-04 15:36:43.895651 EDT,413 2023-10-04 15:36:43.931145 EDT,689 , Trace Name,SignalName3 Signal, Timestamp,Value 2023-10-04 15:36:43.757193 EDT,9867 2023-10-04 15:36:43.829083 EDT,1257 2023-10-04 15:36:43.895651 EDT,5736 2023-10-04 15:36:43.931145 EDT,4935
目标数据格式
我们的目标是将上述堆叠数据转换为一个宽格式的DataFrame，其中 Timestamp 列作为主键，每个 SignalName 作为一个独立的列，包含其对应的值。
Timestamp SignalName1 SignalName2 SignalName3 0 2023-10-04 15:36:43 13 131 9867 1 2023-10-04 15:36:43 14 238 1257 2 2023-10-04 15:36:43 17 413 5736 3 2023-10-04 15:36:43 11 689 4935
解决方案概述
解决这类问题的核心思路是“分而治之”：
- 分块读取： 首先，将整个文件内容读取为字符串，然后使用正则表达式根据空行分隔符将字符串分割成多个独立的子字符串，每个子字符串代表一个数据块。
- 逐块解析： 对每个数据块字符串，使用 pandas.read_csv 进行解析。在解析时，需要跳过块头部的元数据行，并正确识别包含信号名称的行作为列名。
- 横向合并： 将所有解析后的DataFrame按照时间戳（或共同的索引）进行横向合并，形成最终的宽格式DataFrame。
实现步骤详解
以下是实现上述解决方案的具体代码和详细解释。
1. 导入必要的库
import pandas as pd import re import io
- pandas：用于数据处理和DataFrame操作。
- re：用于正则表达式，以便于分割文件内容。
- io：用于将字符串模拟成文件对象，供 pd.read_csv 读取。
2. 读取文件并分块
首先，我们需要打开CSV文件，读取其全部内容，然后使用正则表达式 re.split 来分割这些内容。
file_path = 'your_stacked_data.csv' # 替换为你的文件路径
with open(file_path, 'r') as f:
file_content = f.read()
# 使用正则表达式分割文件内容
# (?:\n,)+\n 匹配一个或多个 "\n," 后面跟着一个 "\n"
# 这是因为空行在CSV中通常表示为 ",<换行符>"
chunks = re.split(r'(?:\n,)+\n', file_content)这里的正则表达式 r'(?:\n,)+\n' 是关键。它匹配一个或多个 , 后面跟着换行符的序列，再以一个换行符结束。这有效地识别了数据块之间的空行分隔符。?: 表示一个非捕获组。
3. 逐块解析数据
遍历 chunks 列表中的每个数据块。对于每个块：
- 使用 io.StringIO() 将字符串数据块转换为一个文件状对象，这样 pd.read_csv 就可以像读取文件一样读取它。
- pd.read_csv() 参数设置：
- header=0：指定第一行（Trace Name,SignalNameX）作为列名。
- skiprows=[1, 2]：跳过第二行（Signal,
）和第三行（Timestamp,Value），因为它们是元数据，且实际的数据列名在第一行已经获取。
- .set_index('Trace Name')：将 Trace Name 列设置为索引。在原始数据块中，这一行实际上是 Trace Name,SignalName1，SignalName1 是其值。通过 header=0，Trace Name 成为列名，而 SignalName1 成为其下方的唯一数据点。将它设置为索引后，它就成了这个数据块的唯一标识符。
dataframes = []
for chunk in chunks:
if chunk.strip(): # 确保块不为空
# 将字符串块转换为文件对象，供pd.read_csv读取
chunk_io = io.StringIO(chunk)
# 读取CSV块，header=0表示第一行是列名，skiprows跳过元数据行
df_chunk = pd.read_csv(chunk_io, header=0, skiprows=[1, 2])
# 将'Trace Name'列设置为索引，其值（如'SignalName1'）将成为索引名
df_chunk = df_chunk.set_index('Trace Name')
dataframes.append(df_chunk)4. 合并所有数据块
现在我们有了一个包含多个DataFrame的列表，每个DataFrame代表一个信号的数据。我们需要将它们横向合并。
# 使用pd.concat沿列方向合并所有DataFrame
# axis=1表示横向合并，默认会根据索引对齐
final_df = pd.concat(dataframes, axis=1)
# 将索引名称从'Trace Name'重命名为'Timestamp'
# 因为合并后，原来的索引（例如'2023-10-04 15:36:43.757193 EDT'）现在代表时间戳
final_df = final_df.rename_axis('Timestamp')
# 重置索引，将'Timestamp'从索引转换为普通列
final_df = final_df.reset_index()pd.concat(dataframes, axis=1) 会根据它们的索引（此时是时间戳）自动对齐并合并。rename_axis('Timestamp') 将原有的索引名称（默认为 None 或前一个DataFrame的索引名）更改为 Timestamp，使其更具语义。最后，reset_index() 将 Timestamp 从索引转换为DataFrame的常规列。
完整代码示例
import pandas as pd
import re
import io
def reshape_stacked_data(file_path):
"""
将堆叠式CSV文件重塑为规范的DataFrame。
Args:
file_path (str): 输入CSV文件的路径。
Returns:
pd.DataFrame: 重塑后的DataFrame。
"""
with open(file_path, 'r') as f:
file_content = f.read()
# 使用正则表达式分割文件内容，根据空行（,后跟换行符）进行分割
# `(?:\n,)+\n` 匹配一个或多个 "\n," 后面跟着一个 "\n"
chunks = re.split(r'(?:\n,)+\n', file_content)
dataframes = []
for chunk in chunks:
if chunk.strip(): # 确保块不为空
# 将字符串块转换为文件对象，供pd.read_csv读取
chunk_io = io.StringIO(chunk)
# 读取CSV块
# header=0: 'Trace Name,SignalNameX' 这一行作为列头
# skiprows=[1, 2]: 跳过 'Signal,' 和 'Timestamp,Value' 两行
df_chunk = pd.read_csv(chunk_io, header=0, skiprows=[1, 2])
# 将'Trace Name'列设置为索引。
# 此时，'Trace Name'列的值（如'SignalName1'）将成为DataFrame的唯一列名，
# 而其数据是时间戳和对应的值。
# 实际上，这里我们期望的是将时间戳作为索引，信号名称作为列。
# 原始答案中的set_index('Trace Name')是巧妙之处，它将时间戳作为数据，
# 而'Trace Name'行的第二个元素（SignalNameX）作为列名。
# 为了达到目标，我们需要将时间戳作为索引。
# 修正：原始逻辑是将'Trace Name'列的值作为新的列名，并将时间戳作为索引。
# 让我们仔细看原始答案的意图：
# pd.read_csv(io.StringIO(chunk), header=0, skiprows=[1,2]).set_index('Trace Name')
# 这一步会产生一个DataFrame，其索引是时间戳，列名是SignalNameX。
# 例如对于SignalName1的块：
# Trace Name
# 2023-10-04 15:36:43.757193 EDT 13
# ...
# 这里的'Trace Name'是列名，而'SignalName1'是这个列的唯一值。
# set_index('Trace Name')后，这个DataFrame的结构是：
# SignalName1
# 2023-10-04 15:36:43.757193 EDT 13
# 2023-10-04 15:36:43.829083 EDT 14
# ...
# 这样是正确的，因为'Trace Name'行的第二个元素就是我们想要的列名。
# 重新执行原始答案的逻辑
df_processed_chunk = pd.read_csv(io.StringIO(chunk), header=0, skiprows=[1,2])
# 获取信号名称，它在第一行的第二列
signal_name = df_processed_chunk.columns[1]
# 将第一列重命名为'Timestamp'
df_processed_chunk.columns = ['Timestamp', signal_name]
# 设置Timestamp为索引
df_processed_chunk = df_processed_chunk.set_index('Timestamp')
dataframes.append(df_processed_chunk)
# 合并所有DataFrame，axis=1表示按列合并，会根据索引（Timestamp）自动对齐
final_df = pd.concat(dataframes, axis=1)
# 将索引名称从默认（或前一个df的索引名）重命名为'Timestamp'
final_df = final_df.rename_axis('Timestamp')
# 重置索引，将'Timestamp'从索引转换为普通列
final_df = final_df.reset_index()
# 格式化时间戳，可选
final_df['Timestamp'] = pd.to_datetime(final_df['Timestamp']).dt.strftime('%Y-%m-%d %H:%M:%S')
return final_df
# 假设你的数据保存在 'stacked_data.csv'
# 创建一个示例文件用于测试
csv_content = """Trace Name,SignalName1
Signal,
Timestamp,Value
2023-10-04 15:36:43.757193 EDT,13
2023-10-04 15:36:43.829083 EDT,14
2023-10-04 15:36:43.895651 EDT,17
2023-10-04 15:36:43.931145 EDT,11
,
Trace Name,SignalName2
Signal,
Timestamp,Value
2023-10-04 15:36:43.757193 EDT,131
2023-10-04 15:36:43.829083 EDT,238
2023-10-04 15:36:43.895651 EDT,413
2023-10-04 15:36:43.931145 EDT,689
,
Trace Name,SignalName3
Signal,
Timestamp,Value
2023-10-04 15:36:43.757193 EDT,9867
2023-10-04 15:36:43.829083 EDT,1257
2023-10-04 15:36:43.895651 EDT,5736
2023-10-04 15:36:43.931145 EDT,4935
"""
with open('stacked_data.csv', 'w') as f:
f.write(csv_content)
# 调用函数并打印结果
output_df = reshape_stacked_data('stacked_data.csv')
print(output_df) 关键代码解析
-
re.split(r'(?:\n,)+\n', file_content):
- 这是将整个文件内容分割成独立数据块的核心。
- \n, 匹配换行符后紧跟一个逗号，这通常是CSV中空行的表示。
- + 表示匹配一个或多个这样的序列。
- \n 确保匹配到空行结束。
- (?:...) 是一个非捕获组，意味着我们只关心匹配模式，不希望它作为分割结果的一部分被返回。
-
io.StringIO(chunk):
- pandas.read_csv 函数通常接受文件路径或文件对象。当我们的数据是内存中的字符串时，io.StringIO 允许我们将其包装成一个行为类似文件的对象，这样 read_csv 就可以直接处理它。
-
pd.read_csv(chunk_io, header=0, skiprows=[1, 2]):
- header=0：指定数据块的第一行（Trace Name,SignalNameX）作为列名。
- skiprows=[1, 2]：跳过紧随其后的两行（Signal,
和 Timestamp,Value），因为它们是元数据或实际数据的标题，而我们已经通过 header=0 获取了所需的列名。
-
df_chunk.columns = ['Timestamp', signal_name] 和 df_chunk = df_chunk.set_index('Timestamp'):
- 在每个数据块内部，pd.read_csv 读取后，第一列是时间戳，第二列是值。
- signal_name = df_processed_chunk.columns[1] 提取了原始第一行第二列的信号名称（例如 SignalName1）。
- 我们将列重命名为 ['Timestamp', 'SignalNameX']。
- 然后将 Timestamp 列设置为索引，这样每个数据块就拥有了以时间戳为索引、以信号名为列的DataFrame。
-
pd.concat(dataframes, axis=1):
- 这是将所有单独处理的DataFrame合并成一个的主要步骤。
- axis=1 表示按列（水平方向）合并。Pandas会根据DataFrame的索引（此时都是时间戳）自动对齐数据。如果时间戳不完全一致，未匹配的单元格将填充 NaN。
-
.rename_axis('Timestamp').reset_index():
- rename_axis('Timestamp')：将合并后DataFrame的索引名称从默认值（或前一个DataFrame的索引名）更改为 Timestamp，使其更具可读性。
- reset_index()：将当前的索引（Timestamp）转换为DataFrame的常规列，并生成一个新的默认整数索引。这使得 Timestamp 成为一个普通的数据列，符合我们最终输出的格式要求。
注意事项
- 文件格式的稳定性： 此方法高度依赖于输入CSV文件格式的稳定性，特别是每个数据块的头部行数和分隔符的模式。如果这些模式发生变化，正则表达式和 skiprows 参数需要相应调整。
- 时间戳格式： 如果时间戳字符串包含时区信息（如 EDT），pd.to_datetime 可以自动解析。如果需要更精细的控制或特定输出格式，可以使用 strftime 进行格式化。
- 内存消耗： 对于非常大的文件，一次性 f.read() 可能会消耗大量内存。在这种情况下，可以考虑逐行读取并动态构建块，或者使用 pd.read_csv 的 chunksize 参数，但那需要更复杂的逻辑来处理非标准分隔。对于本文描述的场景，文件通常不会大到无法一次性读入内存。
- 错误处理： 实际应用中，应增加错误处理机制，例如当文件不存在、数据块格式不正确或 pd.read_csv 解析失败时。
总结
通过结合使用Python的 re 模块进行字符串分割和Pandas强大的数据处理能力，我们可以有效地将复杂的、堆叠式的CSV数据重塑为结构清晰、易于分析的DataFrame。这种方法灵活且高效，为处理非标准数据格式提供了一个通用的解决方案，极大地提升了数据分析的效率和便利性。理解每个步骤背后的逻辑和Pandas函数的用途是掌握此技术的关键。
