2. Preprocesamiento - Lending Club · Aprendizaje automático

Este capítulo corresponde a la sección 9.10.4.2 del proyecto. Aplica las decisiones del resumen ejecutivo del EDA (sección 1.6) de forma idéntica en scikit-learn y en PySpark, partiendo en ambos casos del CSV original completo:

Se genera una única partición 80/20 estratificada y se guarda en disco. Los dos entornos la leen.
Se ajustan los transformadores solo con el conjunto de entrenamiento y después se aplican al de prueba.
Se verifica numéricamente que los dos entornos producen exactamente las mismas variables.

El código de las transformaciones está en lc_utils.py (SkPreprocessor y spark_preprocess), para que los capítulos de modelado usen exactamente el mismo preprocesamiento que se valida aquí.

import json
import os
import platform
import time
import warnings

import numpy as np
import pandas as pd
import psutil

import lc_utils as U

warnings.filterwarnings("ignore")
pd.set_option("display.max_columns", 40)
pd.set_option("display.width", 200)

2.1 Partición común de los datos¶

La prueba de DeLong compara los AUC calculados sobre las mismas observaciones, así que scikit-learn y PySpark deben evaluar exactamente el mismo conjunto de prueba. La partición se define una sola vez:

con train_test_split sobre el identificador id;
con test_size=0.2, stratify=default y random_state=42;
y se guarda en data/split.parquet con las columnas id y split.

PySpark no usa randomSplit: lee este archivo y separa los conjuntos filtrando por split.

t0 = time.time()
df = U.load_model_frame()          # CSV completo -> filtro de desenlace -> variables del modelo
t_sk_load = time.time() - t0

split = U.make_split(df)
split.to_parquet(U.SPLIT_PARQUET, index=False)
df = df.merge(split, on="id", validate="one_to_one")
t_sk_split = time.time() - t0 - t_sk_load

print(f"Filas: {len(df):,} | lectura y construcción de variables: {t_sk_load:.1f} s | partición: {t_sk_split:.1f} s")
print("Archivo:", U.SPLIT_PARQUET)

Filas: 1,345,310 | lectura y construcción de variables: 31.1 s | partición: 4.9 s
Archivo: C:\Users\User\Desktop\GEOLOGÍA\MAESTRÍA\Semestre 2\MachineLearning\jbook_ml202330\data\split.parquet

part = df.groupby("split")[U.TARGET].agg(n="size", defaults="sum", pct_default="mean")
part["pct_default"] *= 100
part["% de filas"] = 100 * part["n"] / part["n"].sum()
print("ids únicos:", split["id"].is_unique,
      "| intersección train/test:", len(set(split.loc[split.split == "train", "id"]) & set(split.loc[split.split == "test", "id"])))
part

ids únicos: True | intersección train/test: 0

La tasa de default es prácticamente la misma en los dos conjuntos, como corresponde a una partición estratificada. El conjunto de prueba tiene 269.062 préstamos: es el tamaño de muestra sobre el que se calculan después DeLong, McNemar y el bootstrap pareado.

2.2 Preprocesamiento con scikit-learn¶

Paso	Variables	Transformación	Parámetro aprendido en train
Selección	16 numéricas y 7 categóricas	se excluyen `fico_range_low`, `installment`, `grade` (redundantes), `mths_since_last_delinq` (50 % de nulos) e `issue_d`	—
Indicadores	`has_delinq_history`, `emp_length_missing`, `mort_acc_missing`	1 si hay historial de mora o si el dato falta	—
Agrupación	`home_ownership`, `purpose`	`ANY`/`NONE`/`OTHER` → `OTHER`; categorías con menos del 1 % → `other`	—
Recorte	`dti`, `revol_util`	recorte (clip) a los percentiles 0,1 y 99,9	percentiles
Transformación	`annual_inc`, `revol_bal`	`log1p`	—
Imputación	`emp_length`, `mort_acc`, `revol_util`, `pub_rec_bankruptcies`, `dti`, `inq_last_6mths`	mediana	medianas
Escalado	16 numéricas y 3 indicadores	`StandardScaler`	media y desviación
Codificación	7 categóricas	`OneHotEncoder(handle_unknown="ignore")`, una columna por categoría	categorías

Los percentiles y las medianas se calculan como un elemento de la muestra (method="inverted_cdf"), que es la misma definición que usa Spark con approxQuantile(..., relativeError=0). Así los dos entornos aprenden exactamente los mismos valores.

Las variables numéricas se escalan para todos los modelos. La regresión logística, LinearSVC y Naive Bayes gaussiano lo necesitan. Los árboles son invariantes a transformaciones monótonas, así que el escalado no les cambia nada. Usar una sola matriz para los seis modelos simplifica la comparación.

Nota sobre la validación cruzada. El enunciado exige GridSearchCV sin Pipeline. Por eso el preprocesamiento se ajusta con todo el conjunto de entrenamiento antes de la búsqueda, y los pliegues de validación comparten medianas, percentiles y medias con los de ajuste. El efecto es despreciable: son pocos parámetros estimados con más de un millón de filas. Además, afecta igual a los dos entornos, porque el CrossValidator de PySpark recibe también los datos ya transformados. El conjunto de prueba no participa en ningún ajuste.

train_df = df[df["split"] == "train"]
test_df = df[df["split"] == "test"]

t0 = time.time()
prep = U.SkPreprocessor().fit(train_df)
X_train, X_test = prep.transform(train_df), prep.transform(test_df)
y_train, y_test = train_df[U.TARGET].to_numpy(), test_df[U.TARGET].to_numpy()
t_sk_fit = time.time() - t0
t_sk_total = t_sk_load + t_sk_split + t_sk_fit

print(f"X_train: {X_train.shape} | X_test: {X_test.shape} | dtype: {X_train.dtype}")
print(f"Memoria de X_train: {X_train.nbytes / 1e6:.0f} MB | ajuste y transformación: {t_sk_fit:.1f} s")

X_train: (1076248, 91) | X_test: (269062, 91) | dtype: float32
Memoria de X_train: 392 MB | ajuste y transformación: 3.6 s

pd.DataFrame(X_train[:5], columns=prep.feature_names_).iloc[:, :24].round(3)

learned_sk = pd.DataFrame({
    "parámetro": [f"clip {c} [p0,1 ; p99,9]" for c in U.CLIP_COLS] + [f"mediana {c}" for c in U.IMPUTE_COLS],
    "scikit-learn": [str(prep.clip_[c]) for c in U.CLIP_COLS] + [prep.median_[c] for c in U.IMPUTE_COLS],
})
ohe_sizes = {c: len(cats) for c, cats in zip(U.MODEL_CAT, prep.ohe_.categories_)}
print("Dummies por variable categórica:", ohe_sizes, "| total de variables:", len(prep.feature_names_))
learned_sk

Dummies por variable categórica: {'term': 2, 'home_ownership': 4, 'verification_status': 3, 'purpose': 8, 'addr_state': 51, 'initial_list_status': 2, 'application_type': 2} | total de variables: 91

# Se guardan los resúmenes que se comparan con PySpark y se libera la memoria antes de iniciar Spark.
sk_summary = {
    "n": {"train": len(train_df), "test": len(test_df)},
    "pct_default": {"train": 100 * y_train.mean(), "test": 100 * y_test.mean()},
    "mean": dict(zip(prep.feature_names_, X_train.mean(axis=0, dtype=np.float64))),
    "scaler_mean": dict(zip(U.SCALED_COLS, prep.scaler_.mean_)),
    "scaler_std": dict(zip(U.SCALED_COLS, prep.scaler_.scale_)),
}
del df, train_df, test_df, X_train, X_test, y_train, y_test

2.3 Preprocesamiento con PySpark¶

La SparkSession se crea con la configuración obligatoria del enunciado (U.spark_session). El flujo de U.spark_preprocess es:

Leer el CSV completo con spark.read.csv(..., escape='"') (ver sección 1.1), filtrar Fully Paid/Charged Off y construir con funciones de Spark SQL las mismas variables que en pandas: años de empleo, antigüedad crediticia, indicadores y agrupaciones.
Unir con la partición común (data/split.parquet) por id y separar train y test filtrando la columna split.
Ajustar con train un Pipeline con estas etapas:
- approxQuantile(…, 0.0) para los límites de recorte (el resultado son 4 números);
- Imputer(strategy="median", relativeError=0.0);
- StringIndexer(handleInvalid="keep") con OneHotEncoder(dropLast=True);
- VectorAssembler y StandardScaler(withMean=True, withStd=True) sobre las numéricas;
- un VectorAssembler final con las dummies.
Transformar train y test y cachearlos con .persist(StorageLevel.MEMORY_AND_DISK) justo después del VectorAssembler, como exige el enunciado.

Durante todo el proceso los datos permanecen como DataFrame de Spark: no se usa .toPandas() ni .collect() sobre el dataset. Al driver solo llegan resultados agregados de tamaño fijo (cuantiles, medianas, conteos y el vector de medias de la sección 2.4).

spark = U.spark_session()
print("Spark", spark.version, "| master:", spark.sparkContext.master)
print({k: v for k, v in spark.sparkContext.getConf().getAll()
       if k in {"spark.sql.shuffle.partitions", "spark.default.parallelism", "spark.executor.memory",
                "spark.driver.memory", "spark.memory.fraction", "spark.memory.storageFraction"}})

Spark 3.5.5 | master: local[*]
{'spark.memory.fraction': '0.8', 'spark.default.parallelism': '400', 'spark.sql.shuffle.partitions': '400', 'spark.driver.memory': '8g', 'spark.executor.memory': '8g', 'spark.memory.storageFraction': '0.3'}

t0 = time.time()
res = U.spark_preprocess(spark)
t_spark_total = time.time() - t0
train_s, test_s, model_s = res["train"], res["test"], res["model"]
print(f"train: {res['n_train']:,} | test: {res['n_test']:,} | tiempo total (lectura + pipeline + caché): {t_spark_total:.1f} s")
print("Etapas del Pipeline:", [type(s).__name__ for s in model_s.stages])

train: 1,076,248 | test: 269,062 | tiempo total (lectura + pipeline + caché): 158.1 s
Etapas del Pipeline: ['ImputerModel', 'StringIndexerModel', 'StringIndexerModel', 'StringIndexerModel', 'StringIndexerModel', 'StringIndexerModel', 'StringIndexerModel', 'StringIndexerModel', 'OneHotEncoderModel', 'VectorAssembler', 'StandardScalerModel', 'VectorAssembler']

print("train cacheado:", train_s.is_cached, "|", train_s.storageLevel)
print("test  cacheado:", test_s.is_cached, "|", test_s.storageLevel)
train_s.printSchema()
train_s.show(3, truncate=110)

train cacheado: True | Disk Memory Serialized 1x Replicated
test  cacheado: True | Disk Memory Serialized 1x Replicated
root
 |-- id: long (nullable = true)
 |-- default: integer (nullable = false)
 |-- features: vector (nullable = true)

+--------+-------+--------------------------------------------------------------------------------------------------------------+
|      id|default|                                                                                                      features|
+--------+-------+--------------------------------------------------------------------------------------------------------------+
|68407277|      0|(91,[0,1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,21,25,30,74,88,89],[-1.2410524863584014,0.1568214644...|
|68341763|      0|(91,[0,1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,20,21,25,31,50,88,90],[0.6398607239337363,-0.5159979584...|
|68426831|      0|(91,[0,1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,24,26,30,46,88,89],[-0.2833924067279532,0.0415408779...|
+--------+-------+--------------------------------------------------------------------------------------------------------------+
only showing top 3 rows

from pyspark.sql import functions as F

rates = {}
for name, d in [("train", train_s), ("test", test_s)]:
    r = d.agg(F.count("*").alias("n"), F.sum(U.TARGET).alias("defaults")).first()
    rates[name] = (r["n"], r["defaults"])
pd.DataFrame(rates, index=["n", "defaults"]).T.assign(pct_default=lambda t: 100 * t["defaults"] / t["n"])

2.4 Verificación de la equivalencia entre entornos¶

Se comparan los parámetros aprendidos (límites de recorte, medianas, media y desviación del escalador) y la media de cada una de las 91 variables resultantes en el conjunto de entrenamiento. En Spark, la media se calcula con Summarizer, que devuelve un único vector agregado.

from pyspark.ml.stat import Summarizer

imp = model_s.stages[0].surrogateDF.first().asDict()
scaler_s = model_s.stages[-2]
rows = []
for c in U.CLIP_COLS:
    for k, q in enumerate(["p0,1", "p99,9"]):
        rows.append({"parámetro": f"clip {c} {q}", "scikit-learn": prep.clip_[c][k], "PySpark": res["bounds"][c][k]})
for c in U.IMPUTE_COLS:
    rows.append({"parámetro": f"mediana {c}", "scikit-learn": prep.median_[c], "PySpark": imp[c]})
for i, c in enumerate(U.SCALED_COLS):
    rows.append({"parámetro": f"media {c}", "scikit-learn": sk_summary["scaler_mean"][c], "PySpark": scaler_s.mean[i]})
    rows.append({"parámetro": f"desv. {c}", "scikit-learn": sk_summary["scaler_std"][c], "PySpark": scaler_s.std[i]})
params = pd.DataFrame(rows).set_index("parámetro")
params["Δ relativa"] = (params["PySpark"] - params["scikit-learn"]).abs() / params["scikit-learn"].abs().clip(lower=1e-12)
with pd.option_context("display.float_format", lambda v: f"{v:,.6g}"):
    display(params.sort_values("Δ relativa", ascending=False).head(12))
print("Máxima diferencia relativa:", params["Δ relativa"].max())

Máxima diferencia relativa: 4.6457726208361726e-07

Los límites de recorte, las medianas y las medias coinciden exactamente. La única diferencia está en la desviación estándar del escalador y es del orden de $5 \times 10^{-7}$ . Se debe a que Spark usa la desviación muestral ( $n-1$ ) y scikit-learn la poblacional ( $n$ ): el cociente entre ambas es $\sqrt{n/(n-1)} \approx 1 + 4{,}6 \times 10^{-7}$ con $n = 1.076.248$ . No tiene efecto práctico en los modelos.

names_s = U.spark_feature_names(train_s)
mean_s = train_s.select(Summarizer.mean(F.col("features")).alias("m")).first()["m"].toArray()
cmp = pd.DataFrame({"scikit-learn": pd.Series(sk_summary["mean"]), "PySpark": pd.Series(mean_s, index=names_s)})
cmp["|Δ|"] = (cmp["PySpark"] - cmp["scikit-learn"]).abs()
print(f"Variables: scikit-learn {len(sk_summary['mean'])} | PySpark {len(names_s)} | mismos nombres: {set(names_s) == set(sk_summary['mean'])}")
print(f"Máxima |Δ| en la media de las variables transformadas (train): {cmp['|Δ|'].max():.2e}")
with pd.option_context("display.float_format", lambda v: f"{v:,.6f}"):
    display(cmp.loc[[n for n in cmp.index if not n.startswith("addr_state")]])

Variables: scikit-learn 91 | PySpark 91 | mismos nombres: True
Máxima |Δ| en la media de las variables transformadas (train): 1.71e-08

Los dos entornos producen las mismas 91 variables, con las mismas categorías, y medias idénticas hasta el orden de 10^-8 (error de redondeo numérico). También coinciden el número de filas y la tasa de default de cada conjunto. Por lo tanto, cualquier diferencia en los resultados de los capítulos siguientes se deberá a los algoritmos de cada librería y no a los datos que reciben.

La tabla omite las 51 dummies de addr_state para abreviar; están incluidas en la comparación del máximo $|\Delta|$ .

2.5 Tiempos de preprocesamiento y hardware¶

vm = psutil.virtual_memory()
hardware = {
    "procesador": platform.processor(),
    "núcleos físicos": psutil.cpu_count(logical=False),
    "núcleos lógicos": psutil.cpu_count(logical=True),
    "RAM total (GB)": round(vm.total / 1e9, 1),
    "sistema": f"{platform.system()} {platform.release()}",
    "Python": platform.python_version(),
    "Spark": spark.version,
}
times = pd.DataFrame({
    "entorno": ["scikit-learn", "PySpark"],
    "lectura CSV + variables (s)": [t_sk_load, np.nan],
    "partición (s)": [t_sk_split, np.nan],
    "ajuste + transformación (s)": [t_sk_fit, np.nan],
    "total (s)": [t_sk_total, t_spark_total],
})
display(pd.Series(hardware, name="valor").to_frame())
times

En PySpark las etapas no se miden por separado. Como la evaluación es perezosa, la lectura del CSV, la unión con la partición, el ajuste del Pipeline y el llenado del caché se ejecutan en los mismos trabajos. El total de Spark incluye varias pasadas completas sobre los datos:

los cuantiles exactos;
las medianas exactas del Imputer;
los siete StringIndexer;
la media y la desviación del StandardScaler;
la materialización del caché.

pandas, en cambio, resuelve cada paso con operaciones vectorizadas en memoria. Con 1,3 millones de filas, que caben en la RAM de un solo equipo, esa diferencia favorece a scikit-learn. Se retoma en la reflexión final.

timings_path = U.DATA_DIR / "timings.json"
timings = json.loads(timings_path.read_text(encoding="utf-8")) if timings_path.exists() else {}
timings["preprocesamiento"] = {"scikit-learn": t_sk_total, "PySpark": t_spark_total}
timings["hardware"] = hardware
timings_path.write_text(json.dumps(timings, indent=2, ensure_ascii=False), encoding="utf-8")

train_s.unpersist()
test_s.unpersist()
spark.stop()
print("Guardado:", timings_path)

Guardado: C:\Users\User\Desktop\GEOLOGÍA\MAESTRÍA\Semestre 2\MachineLearning\jbook_ml202330\data\timings.json