Python中的数据质量（Data Quality）保障：实现数据校验、清洗与漂移检测

Python中的数据质量（Data Quality）保障：实现数据校验、清洗与漂移检测

大家好，今天我们来深入探讨Python在数据质量保障中的应用。数据质量是数据分析、机器学习等一切数据驱动型任务的基石。劣质的数据会导致错误的结论、糟糕的决策，最终影响业务成果。因此，构建一套健全的数据质量保障体系至关重要。 本次讲座将涵盖数据校验、数据清洗和数据漂移检测三个核心方面，并结合实际代码示例，展示如何利用Python提升数据质量。

一、数据校验（Data Validation）

数据校验是指检查数据是否符合预定义的规则和约束。目的是尽早发现错误，防止脏数据进入后续处理流程。常见的数据校验类型包括：

• 类型校验: 检查数据是否为期望的数据类型（例如，整数、字符串、日期）。
• 范围校验: 检查数值数据是否在可接受的范围内。
• 格式校验: 检查数据是否符合特定的格式（例如，电子邮件地址、电话号码）。
• 唯一性校验: 检查数据是否唯一。
• 完整性校验: 检查是否存在缺失值。
• 一致性校验: 检查相关数据之间是否一致。
• 自定义校验: 根据业务规则进行校验。

下面我们通过Python代码演示几种常见的校验方法：

1. 类型校验

def validate_type(data, expected_type):
  """
  验证数据类型。

  Args:
    data: 要验证的数据。
    expected_type: 期望的数据类型。

  Returns:
    True 如果数据类型正确，否则 False。
  """
  return isinstance(data, expected_type)

# 示例
age = 30
name = "Alice"
is_valid_age = validate_type(age, int)
is_valid_name = validate_type(name, str)

print(f"Age is valid: {is_valid_age}")
print(f"Name is valid: {is_valid_name}")

2. 范围校验

def validate_range(data, min_value, max_value):
  """
  验证数据是否在指定范围内。

  Args:
    data: 要验证的数据。
    min_value: 最小值。
    max_value: 最大值。

  Returns:
    True 如果数据在范围内，否则 False。
  """
  return min_value <= data <= max_value

# 示例
temperature = 25
is_valid_temperature = validate_range(temperature, -40, 50)

print(f"Temperature is valid: {is_valid_temperature}")

3. 格式校验

import re

def validate_email(email):
  """
  验证电子邮件地址格式。

  Args:
    email: 要验证的电子邮件地址。

  Returns:
    True 如果电子邮件地址格式正确，否则 False。
  """
  pattern = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+.[a-zA-Z]{2,}$"
  return re.match(pattern, email) is not None

# 示例
email = "[email protected]"
is_valid_email = validate_email(email)

print(f"Email is valid: {is_valid_email}")

4. 完整性校验

def validate_missing_values(data):
  """
  检查数据中是否存在缺失值（None或NaN）。

  Args:
    data: 要检查的数据，可以是列表、字典或Pandas DataFrame。

  Returns:
    包含缺失值字段的列表。
  """
  missing_fields = []
  if isinstance(data, list):
    for i, value in enumerate(data):
      if value is None:
        missing_fields.append(f"Index {i}")
  elif isinstance(data, dict):
    for key, value in data.items():
      if value is None:
        missing_fields.append(key)
  elif isinstance(data, pd.DataFrame):
      missing_values = data.isnull().sum()
      for col in missing_values.index:
          if missing_values[col] > 0:
              missing_fields.append(col)

  return missing_fields

# 示例
import pandas as pd

data_list = [1, 2, None, 4]
data_dict = {"name": "Alice", "age": None, "city": "New York"}
data_df = pd.DataFrame({"col1": [1, 2, None], "col2": ["A", None, "C"]})

missing_in_list = validate_missing_values(data_list)
missing_in_dict = validate_missing_values(data_dict)
missing_in_df = validate_missing_values(data_df)

print(f"Missing values in list: {missing_in_list}")
print(f"Missing values in dict: {missing_in_dict}")
print(f"Missing values in DataFrame: {missing_in_df}")

5. 使用Pandas进行数据校验

Pandas库提供了强大的数据处理能力，可以方便地进行数据校验。

import pandas as pd

def validate_data_with_pandas(df, schema):
  """
  使用Pandas和预定义的Schema进行数据校验。

  Args:
    df: Pandas DataFrame。
    schema:  包含列名和数据类型信息的字典，例如：{"column_name": "data_type"}

  Returns:
    一个包含校验错误的DataFrame，如果没有错误则返回一个空的DataFrame。
  """
  error_df = pd.DataFrame()

  for column, data_type in schema.items():
    if column not in df.columns:
      print(f"Error: Column '{column}' not found in DataFrame.")
      continue # 或者抛出异常

    if data_type == 'int':
      try:
        df[column] = pd.to_numeric(df[column], errors='raise')
        df[column] = df[column].astype('int') #显式转换为int, 避免float
      except ValueError:
        error_rows = df[~df[column].astype(str).str.match(r'^-?d+$')] # 找出不能转换为int的行
        error_rows['error_message'] = f"Invalid data type for column '{column}': Expected integer."
        error_df = pd.concat([error_df, error_rows], ignore_index=True)

    elif data_type == 'float':
      try:
        df[column] = pd.to_numeric(df[column], errors='raise') # 尝试转换为数值
      except ValueError:
        error_rows = df[~df[column].astype(str).str.match(r'^-?d*.?d*$')]
        error_rows['error_message'] = f"Invalid data type for column '{column}': Expected float."
        error_df = pd.concat([error_df, error_rows], ignore_index=True)

    elif data_type == 'string':
      df[column] = df[column].astype(str)  # 强制转换为字符串
    elif data_type == 'date':
        try:
            df[column] = pd.to_datetime(df[column], errors='raise')
        except ValueError:
            error_rows = df[pd.to_datetime(df[column], errors='coerce').isnull()] #找出不能转换为datetime的行
            error_rows['error_message'] = f"Invalid data type for column '{column}': Expected date."
            error_df = pd.concat([error_df, error_rows], ignore_index=True)

    # 可以添加更多数据类型和校验规则

  return error_df

# 示例
data = {'id': [1, 2, '3'], 'name': ['Alice', 'Bob', 'Charlie'], 'age': ['30', '35', 'abc'], 'date': ['2023-01-01', '2023-01-02', 'invalid']}
df = pd.DataFrame(data)

schema = {'id': 'int', 'name': 'string', 'age': 'int', 'date': 'date'}
error_df = validate_data_with_pandas(df, schema)

print("Error DataFrame:")
print(error_df)

print("nOriginal DataFrame (after potential type conversions):")
print(df)  # 注意：原始DataFrame可能会因为类型转换而发生改变

这个例子展示了如何使用Pandas进行类型校验。我们定义了一个schema字典，指定了每列的期望数据类型。validate_data_with_pandas函数遍历每一列，并根据schema中的类型进行校验。如果发现类型错误，则将错误信息添加到error_df中。 这里错误处理使用的是try-except 结构，并使用 pd.to_numeric和pd.to_datetime方法尝试转换数据类型，如果转换失败则捕获异常。

表格：数据校验类型与Python实现

校验类型	描述	Python 实现
类型校验	检查数据是否为期望的数据类型。	isinstance()函数
范围校验	检查数值数据是否在可接受的范围内。	比较运算符 (<=, >=)
格式校验	检查数据是否符合特定的格式。	正则表达式 (re模块)
完整性校验	检查是否存在缺失值。	None检查, pandas.isnull()
一致性校验	检查相关数据之间是否一致（例如，订单总额等于商品单价乘以数量）。	自定义逻辑，根据业务规则实现。
唯一性校验	检查数据是否唯一（例如，用户ID）。	pandas.Series.is_unique, pandas.DataFrame.duplicated()
自定义校验	根据业务规则进行校验（例如，年龄必须大于18岁）。	自定义函数，结合业务逻辑实现。

二、数据清洗（Data Cleaning）

数据清洗是指处理数据中的错误、缺失值、重复值和不一致性。目标是生成干净、一致的数据，以便进行准确的分析。常见的数据清洗操作包括：

• 处理缺失值: 填充缺失值或删除包含缺失值的记录。
• 处理重复值: 删除重复的记录。
• 处理异常值: 识别和处理超出正常范围的数据。
• 数据类型转换: 将数据转换为正确的数据类型。
• 字符串处理: 清理和标准化字符串数据（例如，去除空格、统一大小写）。
• 数据标准化/归一化: 将数据缩放到一个特定的范围。

1. 处理缺失值

import pandas as pd

def handle_missing_values(df, strategy="mean", fill_value=None):
  """
  处理DataFrame中的缺失值。

  Args:
    df: Pandas DataFrame。
    strategy: 填充缺失值的策略，可以是 "mean" (均值), "median" (中位数), "mode" (众数), "constant" (常数)。
    fill_value: 如果 strategy 是 "constant"，则使用此值填充缺失值。

  Returns:
    处理后的DataFrame。
  """
  df = df.copy() #避免修改原始dataframe

  for col in df.columns:
    if df[col].isnull().any(): #先检查是否有缺失值，避免不必要的计算
      if strategy == "mean":
        df[col].fillna(df[col].mean(), inplace=True)
      elif strategy == "median":
        df[col].fillna(df[col].median(), inplace=True)
      elif strategy == "mode":
        df[col].fillna(df[col].mode()[0], inplace=True) # mode() 返回 Series, 取第一个值
      elif strategy == "constant":
        if fill_value is None:
          raise ValueError("fill_value must be specified when strategy is 'constant'.")
        df[col].fillna(fill_value, inplace=True)
      else:
        raise ValueError("Invalid strategy. Choose from 'mean', 'median', 'mode', or 'constant'.")
  return df

# 示例
data = {'col1': [1, 2, None, 4], 'col2': [5, None, 7, 8]}
df = pd.DataFrame(data)

df_filled_mean = handle_missing_values(df.copy(), strategy="mean") # 使用均值填充
df_filled_constant = handle_missing_values(df.copy(), strategy="constant", fill_value=0) # 使用0填充

print("Original DataFrame:")
print(df)
print("nDataFrame with missing values filled with mean:")
print(df_filled_mean)
print("nDataFrame with missing values filled with 0:")
print(df_filled_constant)

2. 处理重复值

import pandas as pd

def handle_duplicates(df, subset=None, keep="first"):
  """
  处理DataFrame中的重复值。

  Args:
    df: Pandas DataFrame。
    subset: 用于识别重复值的列的列表。如果为 None，则使用所有列。
    keep:  指定保留哪个重复项。可以是 "first" (保留第一个), "last" (保留最后一个), False (删除所有重复项)。

  Returns:
    处理后的DataFrame。
  """
  df = df.copy()
  df.drop_duplicates(subset=subset, keep=keep, inplace=True)
  return df

# 示例
data = {'col1': [1, 2, 2, 4], 'col2': ['A', 'B', 'B', 'D']}
df = pd.DataFrame(data)

df_no_duplicates = handle_duplicates(df.copy()) # 删除所有重复行
df_no_duplicates_col1 = handle_duplicates(df.copy(), subset=['col1']) #只根据col1删除重复行

print("Original DataFrame:")
print(df)
print("nDataFrame with duplicates removed:")
print(df_no_duplicates)
print("nDataFrame with duplicates removed based on col1:")
print(df_no_duplicates_col1)

3. 处理异常值

import pandas as pd
import numpy as np

def handle_outliers_iqr(df, column, threshold=1.5):
  """
  使用 IQR 方法处理DataFrame中的异常值。

  Args:
    df: Pandas DataFrame。
    column: 要处理的列。
    threshold: IQR 的倍数，用于定义异常值的范围。

  Returns:
    处理后的DataFrame，异常值被替换为 NaN。
  """
  df = df.copy()
  Q1 = df[column].quantile(0.25)
  Q3 = df[column].quantile(0.75)
  IQR = Q3 - Q1
  lower_bound = Q1 - threshold * IQR
  upper_bound = Q3 + threshold * IQR

  # 将超出范围的值替换为NaN
  df[column] = np.where((df[column] < lower_bound) | (df[column] > upper_bound), np.nan, df[column])
  return df

def handle_outliers_zscore(df, column, threshold=3):
    """
    使用Z-score方法处理 DataFrame 中的异常值.

    Args:
        df: Pandas DataFrame.
        column:  要处理的列.
        threshold:  Z-score的阈值，用于定义异常值的范围.

    Returns:
        处理后的 DataFrame，异常值被替换为 NaN.
    """
    df = df.copy()
    mean = df[column].mean()
    std = df[column].std()
    z_scores = np.abs((df[column] - mean) / std)
    df[column] = np.where(z_scores > threshold, np.nan, df[column]) #替换为NaN
    return df

# 示例
data = {'col1': [1, 2, 3, 4, 5, 100]}
df = pd.DataFrame(data)

df_no_outliers_iqr = handle_outliers_iqr(df.copy(), 'col1')
df_no_outliers_zscore = handle_outliers_zscore(df.copy(), 'col1')

print("Original DataFrame:")
print(df)
print("nDataFrame with outliers handled using IQR:")
print(df_no_outliers_iqr)
print("nDataFrame with outliers handled using Z-score:")
print(df_no_outliers_zscore)

4. 数据类型转换

在数据校验的例子中，已经展示了使用 pd.to_numeric 和 astype 进行数据类型转换。这里不再重复。

5. 字符串处理

import pandas as pd

def clean_string_data(df, column):
  """
  清理字符串数据，包括去除空格、统一大小写等。

  Args:
    df: Pandas DataFrame。
    column: 要处理的字符串列。

  Returns:
    处理后的DataFrame。
  """
  df = df.copy()
  df[column] = df[column].str.strip()  # 去除首尾空格
  df[column] = df[column].str.lower()  # 转换为小写
  return df

# 示例
data = {'col1': ['  Alice  ', 'BOB', ' Charlie ']}
df = pd.DataFrame(data)

df_cleaned = clean_string_data(df.copy(), 'col1')

print("Original DataFrame:")
print(df)
print("nCleaned DataFrame:")
print(df_cleaned)

表格：数据清洗操作与Python实现

清洗操作	描述	Python 实现
缺失值处理	填充缺失值或删除包含缺失值的记录。	pandas.DataFrame.fillna(), pandas.DataFrame.dropna()
重复值处理	删除重复的记录。	pandas.DataFrame.drop_duplicates()
异常值处理	识别和处理超出正常范围的数据。	IQR方法（计算四分位数和IQR，然后根据阈值过滤异常值）、Z-score方法（计算Z-score，然后根据阈值过滤异常值）， 可以使用numpy.where()将异常值替换为 NaN。
数据类型转换	将数据转换为正确的数据类型。	pandas.to_numeric(), pandas.to_datetime(), pandas.DataFrame.astype()
字符串处理	清理和标准化字符串数据。	pandas.Series.str.strip(), pandas.Series.str.lower(), pandas.Series.str.upper(), pandas.Series.str.replace()
数据标准化/归一化	将数据缩放到一个特定的范围 (例如 0 到 1)。	可以使用 sklearn.preprocessing.MinMaxScaler (归一化到 0-1 范围) 或者 sklearn.preprocessing.StandardScaler (标准化到均值为 0，方差为 1)

三、数据漂移检测（Data Drift Detection）

数据漂移是指模型输入数据的分布随时间发生变化。这种变化会导致模型性能下降。数据漂移检测的目的是尽早发现数据分布的变化，以便及时采取措施（例如，重新训练模型）。常见的数据漂移检测方法包括：

• 统计检验: 使用统计检验（例如，卡方检验、 Kolmogorov-Smirnov 检验）比较两个数据集的分布。
• 距离度量: 使用距离度量（例如，KL散度、Wasserstein距离）衡量两个数据集的分布差异。
• 模型监控: 监控模型的性能指标，如果性能下降，则可能存在数据漂移。

1. 使用卡方检验进行分类特征的漂移检测

import pandas as pd
from scipy.stats import chi2_contingency

def detect_categorical_drift_chi2(df_reference, df_current, column, alpha=0.05):
  """
  使用卡方检验检测分类特征的漂移。

  Args:
    df_reference: 参考数据集的 Pandas DataFrame。
    df_current: 当前数据集的 Pandas DataFrame。
    column: 要检测的分类特征列。
    alpha: 显著性水平。

  Returns:
    True 如果检测到漂移，否则 False。
  """

  contingency_table = pd.crosstab(df_reference[column], df_current[column])
  chi2, p, dof, expected = chi2_contingency(contingency_table)

  print(f"Chi-square statistic: {chi2}")
  print(f"P-value: {p}")

  return p < alpha

# 示例
data_reference = {'category': ['A', 'B', 'C', 'A', 'B']}
df_reference = pd.DataFrame(data_reference)

data_current = {'category': ['A', 'B', 'C', 'B', 'D']}
df_current = pd.DataFrame(data_current)

drift_detected = detect_categorical_drift_chi2(df_reference, df_current, 'category')

print(f"Drift detected: {drift_detected}")

2. 使用 Kolmogorov-Smirnov 检验进行数值特征的漂移检测

import pandas as pd
from scipy.stats import ks_2samp

def detect_numerical_drift_ks(df_reference, df_current, column, alpha=0.05):
  """
  使用 Kolmogorov-Smirnov 检验检测数值特征的漂移。

  Args:
    df_reference: 参考数据集的 Pandas DataFrame。
    df_current: 当前数据集的 Pandas DataFrame。
    column: 要检测的数值特征列。
    alpha: 显著性水平。

  Returns:
    True 如果检测到漂移，否则 False。
  """
  ks_statistic, p_value = ks_2samp(df_reference[column], df_current[column])

  print(f"KS statistic: {ks_statistic}")
  print(f"P-value: {p_value}")

  return p_value < alpha

# 示例
data_reference = {'value': [1, 2, 3, 4, 5]}
df_reference = pd.DataFrame(data_reference)

data_current = {'value': [2, 3, 4, 5, 6]}
df_current = pd.DataFrame(data_current)

drift_detected = detect_numerical_drift_ks(df_reference, df_current, 'value')

print(f"Drift detected: {drift_detected}")

3. 使用Population Stability Index (PSI)

PSI 是一个衡量两个样本分布相似度的指标，常用于评估模型输入特征的数据漂移情况。PSI 的计算公式如下：

PSI = SUM((实际占比 – 预期占比) * ln(实际占比 / 预期占比))

import pandas as pd
import numpy as np

def calculate_psi(expected, actual, buckettype='bins', n_buckets=10, axis=0):
    """
    Calculates the population stability index (PSI).

    Args:
        expected: Array or pandas Series of expected values.
        actual: Array or pandas Series of actual values.
        buckettype: 'bins' splits into even splits, 'quantiles' splits into quantile buckets.
        n_buckets: Number of buckets to split into.
        axis: Axis to perform calculations along.

    Returns:
        PSI value.
    """

    def sub_psi(e_perc, a_perc):
        '''Calculate single PSI value'''
        if a_perc == 0:
            a_perc = 0.0001
        if e_perc == 0:
            e_perc = 0.0001

        value = (a_perc - e_perc) * np.log(a_perc / e_perc)
        return value

    if buckettype == 'bins':
        buckets = np.linspace(np.min(np.hstack([actual, expected])), np.max(np.hstack([actual, expected])), n_buckets + 1)
    elif buckettype == 'quantiles':
        buckets = np.percentile(np.hstack([actual, expected]), np.linspace(0, 100, n_buckets + 1))
    else:
        raise ValueError('buckettype must be "bins" or "quantiles"')

    expected_percents = np.histogram(expected, buckets=buckets)[0] / len(expected)
    actual_percents = np.histogram(actual, buckets=buckets)[0] / len(actual)

    psi_value = np.sum(sub_psi(expected_percents[i], actual_percents[i]) for i in range(len(expected_percents)))
    return psi_value

# 示例
data_reference = {'value': np.random.normal(0, 1, 1000)}  # 均值为0，标准差为1的正态分布
df_reference = pd.DataFrame(data_reference)

data_current = {'value': np.random.normal(0.5, 1, 1000)} # 均值为0.5，标准差为1的正态分布，发生了漂移
df_current = pd.DataFrame(data_current)

psi = calculate_psi(df_reference['value'], df_current['value'], buckettype='quantiles', n_buckets=10)

print(f"PSI value: {psi}")

#PSI解读：
#PSI < 0.1: No significant change.
#0.1 <= PSI < 0.2: Small shift in population.
#PSI >= 0.2: Significant shift in population.

表格：数据漂移检测方法与Python实现

漂移检测方法	描述	Python 实现
卡方检验	比较两个分类变量的分布是否相同。	scipy.stats.chi2_contingency()
Kolmogorov-Smirnov 检验	比较两个数值变量的分布是否相同。	scipy.stats.ks_2samp()
Population Stability Index (PSI)	衡量两个样本分布的相似度。	自定义函数实现，基于 numpy.histogram() 和 numpy.log()
模型监控	监控模型的性能指标，如果性能下降，则可能存在数据漂移。	需要结合具体的模型和监控工具实现。

总结

这次讲座我们探讨了Python在数据质量保障中的应用，涵盖了数据校验、数据清洗和数据漂移检测三个核心方面。通过代码示例，我们演示了如何使用Python库（例如，Pandas、scipy）实现各种数据质量检查和处理任务。

保障数据质量，让数据分析更有价值

构建一套健全的数据质量保障体系至关重要，它可以确保数据的准确性、完整性和一致性，从而提高数据分析、机器学习等任务的可靠性和有效性。

更多IT精英技术系列讲座，到智猿学院