Python 清洗原始数据

Python 清洗原始数据，EDA中经常需要清洗原始数据，通过执行一系列标量函数，把输入数据转换为可用的数据集合。

下面介绍一个简化的数据集合，这是EDA领域中常用于演示数据处理技术的数据集：安斯库姆四重奏，名字来自F. J. Anscombe于1973年在《美国统计学家》杂志上发表的文章“Graphs in Statistical Analysis”。该数据集的前几行如下所示：

Anscombe's quartet
I  II  III  IV
x  y  x  y  x  y  x  y
10.0  8.04  10.0  9.14       10.0  7.46  8.0  6.58
8.0      6.95  8.0  8.14  8.0  6.77  8.0  5.76
13.0  7.58  13.0  8.74  13.0  12.74  8.0  7.71

csv模块无法直接解析这样的数据格式，需要先解码才能从文件中抽取有用的信息。由于数据之间都是用Tab分隔的，可以使用csv.reader()函数处理每一行数据。首先定义一个数据迭代器，如下所示：

import csv
from typing import IO, Iterator, List, Text, Union, Iterable
def row_iter(source: IO) -> Iterator[List[Text]]:
    return csv.reader(source, delimiter="\t")

这里只是简单地把文件对象包裹在csv.reader()函数中，生成一个行迭代器。类型模块为文件对象提供了方便的定义：IO。csv.reader()负责迭代处理所有行，每一行是一个文本值列表。额外添加一个类型定义Row = List[Text]会使之更明确。

在如下所示的上下文中使用row_iter()函数：

with open("Anscombe.txt") as source:
    print(list(row_iter(source)))

虽然确实包含了有用的数据，但返回结果的前3行并不是数据，如下所示：

[["Anscombe's quartet"],
 ['I', 'II', 'III', 'IV'],
 ['x', 'y', 'x', 'y', 'x', 'y', 'x', 'y'],

需要过滤掉这些非数据的行。下面的函数会移除迭代器的前3行，返回包含剩余行的迭代器。

def head_split_fixed(
        row_iter: Iterator[List[Text]]
    ) -> Iterator[List[Text]]:
    title = next(row_iter)
    assert (len(title) == 1
        and title[0] == "Anscombe's quartet")
    heading = next(row_iter)
    assert (len(heading) == 4
        and heading == ['I', 'II', 'III', 'IV'])
    columns = next(row_iter)
    assert (len(columns) == 8
        and columns == ['x','y', 'x','y', 'x','y', 'x','y'])
    return row_iter

该函数会移除可迭代对象的前3行，并检查剩余每一行是否符合预期，如果不相符，说明文件已损坏，或者这并不是要处理的文件。

由于row_iter()和head_split_fixed()函数都使用可迭代对象作为输入参数，可以将它们合并，如下所示：

with open("Anscombe.txt") as source:
    print(list(head_split_fixed(row_iter(source))))

将一个迭代器的处理结果作为参数传递给另一个迭代器，实际上这是一个复合函数。当然，整个数据清洗并没有到此结束，还需要将字符串转换为浮点数，然后将每行中4个并行序列的数据分开。

使用高阶函数（如map()和filter()）来进行最终的转换和数据抽取会比较简单。