BioPlayground

Volver a la lista

Pipeline para la limpieza de datos experimentales: gestión reproducible de valores faltantes, duplicados y valores atípicos.

Se define un flujo de trabajo explícito en pandas para gestionar valores faltantes, duplicados, valores atípicos y errores de formato en archivos CSV. Se evita el uso de argumentos mutables y se registra cada paso del proceso.

Intermedio
|
80min
|
Verificado (2026-07)
Limpieza de datos.pandas pipelinevalores faltantesEliminación de duplicadosDetectar valores atípicosIQRAnálisis reproducible
Progreso0/12 (0%)

Pipeline de limpieza de datos experimentales: manejo reproducible de valores faltantes, duplicados y atípicos

Al finalizar este tema

Podrás construir un pipeline que purifique de manera reproducible datos experimentales en formato CSV desordenado, integrando conceptos aprendidos en los libros de texto como limpieza de datos, apply/map y la trampa de los argumentos mutables. Esto elimina el problema de obtener resultados no reproducibles debido a limpiezas inconsistentes.

Este texto es un ejemplo genérico con fines educativos. En la práctica, se utilizan herramientas más sofisticadas como dbt, Great Expectations y Pandera.

"¿Por qué no aparecen los resultados de la semana pasada?" — Las trampas de la limpieza manual

Imaginemos que reciben y analizan datos experimentales en formato CSV cada semana. La primera semana:

  1. Abrir en Excel y verificar visualmente las celdas vacías
  2. Eliminar los valores que parecen incorrectos
  3. Combinar los datos provenientes de varios archivos en uno solo
  4. Guardar y analizar

La semana siguiente se repite el mismo proceso. Y entonces tu asesor pregunta: "¿Por qué no se reprodujeron los resultados de la semana pasada esta semana?"

No es que haya una sola causa.

Problema 1: Los valores que se eliminan visualmente varían cada vez. En una semana se eliminan los valores que están fuera de 3σ, y en otra semana, los que están fuera de 4σ.

Problema 2: La forma de manejar los valores faltantes varía cada vez. En una semana se eliminan, y en otra se reemplazan por la media.

Problema 3: No recuerdas qué archivos faltaron esta vez.

El enfoque correcto es un pipeline explícito. Cada paso de limpieza se define como código, y ese código se guarda y se gestiona con control de versiones. Si se vuelve a procesar los mismos datos con el mismo código, se obtendrán exactamente los mismos resultados. Esta es la base de la reproducibilidad científica.

De la caja negra a los componentes

Componente 1: Pasos explícitos en pandas

Cada acción de limpieza como una función separada.

python
import pandas as pd
import numpy as np




def load_and_validate(path: str, required_cols: list[str]) -> pd.DataFrame:
    df = pd.read_csv(path)
    missing = set(required_cols) - set(df.columns)
    if missing:
        raise ValueError(f"Missing columns: {missing}")
    return df




def remove_duplicates(df: pd.DataFrame, subset: list[str]) -> tuple[pd.DataFrame, int]:
    n_before = len(df)
    df_clean = df.drop_duplicates(subset=subset).copy()
    return df_clean, n_before - len(df_clean)




def drop_missing(df: pd.DataFrame, required_cols: list[str]) -> tuple[pd.DataFrame, int]:
    n_before = len(df)
    df_clean = df.dropna(subset=required_cols).copy()
    return df_clean, n_before - len(df_clean)




def filter_outliers_iqr(df: pd.DataFrame, col: str, factor: float = 1.5) -> tuple[pd.DataFrame, int]:
    q1 = df[col].quantile(0.25)
    q3 = df[col].quantile(0.75)
    iqr = q3 - q1
    low, high = q1 - factor * iqr, q3 + factor * iqr
    n_before = len(df)
    df_clean = df[(df[col] >= low) & (df[col] <= high)].copy()
    return df_clean, n_before - len(df_clean)

Punto clave: cada función devuelve cuántos elementos se eliminaron (recuento). Esto se registrará más adelante.

Parte 2: Transformación de columnas con apply y map

La transformación de valores se realiza explícitamente con apply o map.

python
def normalize_gene_symbols(df: pd.DataFrame, col: str = "gene") -> pd.DataFrame:
    """
    Convierte los símbolos génicos al estándar en mayúsculas.
    """
    df = df.copy()
    df[col] = df[col].str.upper().str.strip()
    return df




def standardize_units(df: pd.DataFrame, col: str, unit_col: str) -> pd.DataFrame:
    """
    Combina el valor y la unidad de concentración para normalizarlos a μM.
    """
    df = df.copy()


    def to_micromolar(row):
        value = row[col]
        unit = row[unit_col].lower()
        multiplier = {"nm": 0.001, "μm": 1, "um": 1, "mm": 1000, "m": 1_000_000}
        return value * multiplier.get(unit, np.nan)


    df["concentration_uM"] = df.apply(to_micromolar, axis=1)
    return df

Nota: Siempre df.copy(). Evite modificar el DataFrame original.

Paso 3: Cómo evitar la trampa de los objetos mutables

Un error común entre los principiantes de Python: el uso de valores predeterminados mutables.

python
# Código peligroso
def add_error_log(errors: list = []) -> list:
    errors.append("something wrong")
    return errors




# Primera llamada: ["something wrong"]
# Segunda llamada: ["something wrong", "something wrong"] — ¡inesperado!

Razón: Los valores predeterminados de las funciones de Python se evalúan solo una vez, en el momento de definir la función. La lista [] asignada como valor predeterminado en errors=[] se crea cuando se define la función y después se reutiliza en cada llamada.

python
def add_error_log(errors: list | None = None) -> list:
    if errors is None:
        errors = []
    errors.append("something wrong")
    return errors

DataFrame también presenta el mismo problema.

python
# Peligro
def process(df):
    df["new_col"] = df["old_col"] * 2  # ¡Modifica el df original!
    return df


original = pd.read_csv("data.csv")
processed = process(original)
# original["new_col"] también existe: el original quedó contaminado.




# Seguro
def process(df):
    df = df.copy()
    df["new_col"] = df["old_col"] * 2
    return df

La reproducibilidad se mantiene si cada paso del flujo de trabajo es una función pura (que no modifica la entrada y devuelve una nueva salida).

Ensamblaje del flujo de trabajo

Ahora, combinemos los componentes anteriores en un único flujo de trabajo.

python
from dataclasses import dataclass, field
from datetime import datetime




@dataclass
class CleaningReport:
    input_rows: int = 0
    output_rows: int = 0
    steps: list = field(default_factory=list)


    def add(self, step_name: str, removed: int, details: str = "") -> None:
        self.steps.append({
            "step": step_name,
            "removed": removed,
            "details": details,
            "timestamp": datetime.now().isoformat()
        })




def clean_experiment_data(
    csv_path: str,
    required_cols: list[str] = None,
    outlier_col: str = "value",
    outlier_factor: float = 1.5
) -> tuple[pd.DataFrame, CleaningReport]:
    required_cols = required_cols or ["gene", "value", "unit"]


    df = load_and_validate(csv_path, required_cols)
    report = CleaningReport(input_rows=len(df))


    df = normalize_gene_symbols(df, "gene")
    report.add("normalize_gene_symbols", 0, "uppercased and stripped")


    df, n_dup = remove_duplicates(df, subset=["gene", "value"])
    report.add("remove_duplicates", n_dup)


    df, n_missing = drop_missing(df, required_cols)
    report.add("drop_missing", n_missing)


    df = standardize_units(df, "value", "unit")
    report.add("standardize_units", 0)


    df, n_outlier = filter_outliers_iqr(df, outlier_col, outlier_factor)
    report.add("filter_outliers_iqr", n_outlier, f"factor={outlier_factor}")


    report.output_rows = len(df)
    return df, report

Uso:

python
clean_df, report = clean_experiment_data("raw_data.csv")


print(f"Entrada: {report.input_rows} filas, salida: {report.output_rows} filas")
for step in report.steps:
    print(f"  {step['step']}: -{step['removed']} ({step['details']})")

Resultado esperado:

text
Entrada: 1000 filas, salida: 934 filas
  normalize_gene_symbols: -0 (uppercased and stripped)
  remove_duplicates: -12
  drop_missing: -30
  standardize_units: -0
  filter_outliers_iqr: -24 (factor=1.5)

Desvanecimiento: tres espacios en blanco para completar.

Espacio en blanco 1: Validación de esquema

Valide explícitamente el tipo y el rango de cada columna.

python
def validate_schema(df: pd.DataFrame, schema: dict) -> list[str]:
    """
    schema = {"col_name": {"type": float, "min": 0, "max": 100}}
    Devuelve una lista de mensajes de error de las validaciones fallidas. Una lista vacía indica que todo pasó.
    """
    errors = []


    for col, rules in schema.items():
        if col not in df.columns:
            errors.append(f"Missing column: {col}")
            continue


        # TODO 1: validar el tipo (comprobar df[col].dtype)
        # TODO 2: validar el intervalo min/max (si existe, comparar df[col].min() y df[col].max())
        pass


    return errors

Pista: if not pd.api.types.is_numeric_dtype(df[col]): errors.append(...). Rango: if "min" in rules and df[col].min() < rules["min"]: errors.append(...).

Espacio en blanco 2: Guardar archivo de registro

Guardar CleaningReport como un registro CSV.

python
def save_report(report: CleaningReport, log_path: str) -> None:
    """
    Guarda report.steps como CSV.
    """
    # TODO: convertir a pd.DataFrame(report.steps) y después usar to_csv
    pass

Espacio en blanco 3: Hash de reproducibilidad

Un hash que verifica si el mismo código produce la misma salida para la misma entrada.

python
import hashlib




def compute_output_hash(df: pd.DataFrame) -> str:
    """
    Ordena el contenido del df y después calcula su hash.
    El mismo df debe producir siempre el mismo hash.
    """
    # TODO 1: ordenar el df en un orden estable (por ejemplo, por todas las columnas)
    # TODO 2: convertirlo en texto con to_csv(index=False)
    # TODO 3: devolver hashlib.sha256(bytes).hexdigest()
    pass

Consejo:

python
sorted_df = df.sort_values(list(df.columns)).reset_index(drop=True)
content = sorted_df.to_csv(index=False).encode()
return hashlib.sha256(content).hexdigest()[:16]

Reflexión: Diferencias con una canalización de datos en producción

dbt (data build tool): Estándar para canalizaciones basadas en SQL. Cada paso es un modelo SQL, las dependencias se gestionan automáticamente y las pruebas se integran en la canalización. Estándar en empresas como Airbnb y Netflix.

Great Expectations / Pandera: Framework para la validación de esquemas y la calidad de los datos. Define las expectativas de cada columna de forma declarativa y notifica las violaciones.

Apache Airflow / Prefect: Orquestación de canalizaciones a gran escala. Funciones como la programación, los reintentos y el relleno.

dask / polars: Si tus datos tienen más de 100 GB, utiliza estas herramientas en lugar de pandas. Procesamiento paralelo y evaluación diferida.

Seguimiento del linaje de datos: En producción, se realiza un seguimiento automático de cada salida, indicando de qué entrada proviene y qué transformaciones ha sufrido. Estándar como OpenLineage.

Proyectos de ampliación

1. Panel de Streamlit: Convierte tu canalización de limpieza en una interfaz de usuario web. Los usuarios pueden cargar archivos CSV y visualizar los informes.

2. Integración de Great Expectations: Utiliza los informes estándar de Great Expectations en lugar de tu CleaningReport.

3. Combinación de varios archivos: Al combinar varios archivos CSV, valida automáticamente la coherencia del esquema antes de la combinación.

4. Catálogo de datos: Almacena y busca los metadatos de cada conjunto de datos experimental en SQLite.

Mapa de componentes de este módulo

  • [F] Limpieza de datos: Manejo explícito de valores faltantes, duplicados y valores atípicos. Registra cada paso.
  • [F] apply/map: Enfoque funcional para la transformación de columnas. Comprende axis=1.
  • [F] La trampa de la mutabilidad: Evita los valores predeterminados mutables y las modificaciones "in situ" de DataFrames. Sigue el principio de funciones puras.
  • [W] E/S de archivos: Lectura y escritura de CSV (se proporciona un script completo).

[F] = Implementarás tú mismo / [W] = Se proporciona como código completo.

💬 Preguntas y comentarios

0 comentarios

Puedes publicar sin iniciar sesión. Los comentarios de invitados no pueden editarse ni eliminarse después.

0/2000

Cargando...