用于列组合的数据帧架构验证

Question

我正在使用pandera开发Pandas DataFrame模式验证代码（python），并正在寻找验证DataFrame是否具有列组合的唯一值的最佳方法

原始数据由其他人提供，采用CSV格式。我的代码将CSV加载到Pandas数据框中，然后pandera数据框模式使用X和Y坐标验证数据框是否具有地理坐标系的列。 数据的本质是数据集中的每一行都应该有一个唯一的X，Y坐标

csv文件的一般形式为：
x坐标、y坐标、A值、B值
12.1234、23.2345、27.23、32.84
34.3456、45.4567、21.12、22.32
..
....

使用pandera，我能想到的唯一方法是：

采取多步骤方法：

1. 将csv文件加载到数据帧中
2. 创建一个单列数据框，其中列名为（比如）“coords”，值由csv数据框坐标列的字符串组合生成
3. 使用pandera DataFrameSchema验证coords DataFrame，该架构使用allow_duplicates=False的pandera列检查该列的唯一性
4. 使用自己的pandera模式验证csv数据帧
5. 合并来自两个模式验证的模式错误，并将其作为错误提出

这种方法似乎有点笨重，我正在寻找其他更能利用pandera灵活性的方法

实现上述功能的代码是：

import pandas as pd
import pandera as pa

def get_valid_coord_df(filename):
  df = pd.read_csv(filename)

  schema = DataFrameSchema(columns={
        'x_coord': pa.Column(pa.Float,
                              Check.in_range(0, 100_000)),
        'y_coord': pa.Column(pa.Float,
                              Check.in_range(0, 100_000)),
        'value_A': pa.Column(pa.Float,
                              Check.in_range(0, 20)),
        'value_B': pa.Column(pa.Float,
                              Check.in_range(0, 20))
        },
        strict=True, ordered=True)

  schema.validate(df, lazy=True) #will raise SchemaError

  df_coord = pd.DataFrame(df['x_coord'].astype(str) + ',' + df['x_coord'].astype(str))

  schema_coord = DataFrameSchema(columns={
       'x_coord': pa.Column(pa.String,
                            allow_duplicates=False)})

  schema_coord.validate(df_coord, lazy=True) #will raise SchemaError

  return df

         
 

Answer 1

在这里，您可以使用wide checks访问Check函数arg中的整个数据帧：

import pandera as pa

schema = pa.DataFrameSchema(
    columns={
        "x_coord": pa.Column(pa.Float, pa.Check.in_range(0, 100_000)),
        "y_coord": pa.Column(pa.Float, pa.Check.in_range(0, 100_000)),
        "value_A": pa.Column(pa.Float, pa.Check.in_range(0, 20)),
        "value_B": pa.Column(pa.Float, pa.Check.in_range(0, 20)),
    },
    # checks at the DataFrameSchema level take the dataframe as input
    # (as opposed to the series at the Column level)
    checks=pa.Check(lambda df: ~df[["x_coord", "y_coord"]].duplicated()),
    strict=True,
    ordered=True,
)

如果您发现自己在代码库中更频繁地使用此检查，还可以定义一个check extension，以便您可以在pa.Check命名空间中访问它：

import pandera.extensions as extensions

@extensions.register_check_method(
    statistics=["columns"], supported_types=pd.DataFrame,
)
def unique_columns(pandas_obj, *, columns):
    return ~pandas_obj[columns].duplicated()


schema = pa.DataFrameSchema(
    checks=pa.Check.unique_columns(["x_coord", "y_coord"])
)

schema.validate(
    pd.DataFrame(
        {
            "x_coord": [1.0, 2.0, 3.0],
            "y_coord": [4.0, 5.0, 6.0],
        }
    )
)