列组合的Pandas数据帧模式验证

Question

我正在使用pandra开发Pandas DataFrame Schema验证代码(在python中(，并正在寻找验证DataFrame是否具有列组合的唯一值的最佳方法。

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

csv文件的一般形式为：
x_coord、y_coord、value_A、value_B
12.1234、23.2345、27.23、32.84
34.3456、45.4567、21.12、22.32

。。。。

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

采取多步骤方法：

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

这种方法似乎有点笨拙，我正在寻找其他选项，以更多地利用pandra中的灵活性。

实现上述功能的代码是：

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

在这里，您可以使用广泛的检查来访问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,
)

如果你发现自己在代码库中更频繁地使用这种检查，你也可以定义一个检查扩展，这样你就可以在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],
}
)
)