unirse a una nueva empresa como ingeniero de datos. Heredas bastantes canalizaciones ETL y eres responsable de mantenerlas. ¿Cuáles crees que son los desafíos de tu trabajo?
Normalmente, es posible que se enfrente a los siguientes problemas:
Cambios en el esquema ascendente: los equipos de desarrolladores pueden agregar o eliminar campos, cambiar tipos de datos o cambiar el nombre de columnas. Cuando un esquema de origen cambia inesperadamente, los trabajos de ETL pueden fallar abruptamente. Para empeorar las cosas, la canalización carga silenciosamente valores nulos o corruptos en tablas posteriores. Problemas de calidad de datos: a veces los trabajos de ETL no fallan inmediatamente, por el contrario, se ejecutan y finalizan con un estado exitoso. Sin embargo, los datos cargados son completamente incorrectos y contienen registros duplicados o faltantes. Falta de documentación: los oleoductos heredados pueden tener pocos documentos o los documentos existentes pueden estar desactualizados. Por lo tanto, no está seguro de si están en línea con la lógica empresarial actual. Crecimiento del volumen y picos de rendimiento: el volumen de datos aumenta a medida que crece el negocio. Una canalización ETL optimizada para un conjunto de datos históricos más pequeño puede volverse lenta, detenerse o fallar fácilmente al procesar volúmenes masivos.
Un flujo de trabajo de prueba automatizado puede ayudarle a solucionar los problemas anteriores. ¿Por qué? Porque el flujo de trabajo estructurado puede ayudarle a comprender rápidamente todos los aspectos clave de una canalización ETL: la lógica empresarial, los algoritmos para la transformación de datos, los tipos de datos, todos los problemas de datos que las canalizaciones ETL deben resolver. Los patrones de prueba son reutilizables: no es necesario diseñar un nuevo flujo de trabajo cada vez que hereda una canalización ETL diferente.
En el artículo de hoy, me centraré en las pruebas automatizadas en ingeniería de datos, incluida la configuración del entorno y un flujo de trabajo práctico. Al final, también analizaré cómo el código asistido por IA puede acelerar el flujo de trabajo y mejorar la productividad.
Haga que el medio ambiente funcione
Si crea el flujo de trabajo de prueba automatizado por primera vez, la configuración del entorno puede llevar algún tiempo. Existen diferentes herramientas y flujos para que los ingenieros de datos configuren el entorno de prueba. Pero si sigues los pasos a continuación, el proceso será fácil y fluido.
En primer lugar, sólo necesita instalar 3 cosas: Docker Desktop, VS Code y Dev Containers Extension.
En su flujo de trabajo de prueba, Docker creará entornos de prueba livianos, aislados y repetibles. Le permite poner en marcha una infraestructura de datos simulada (por ejemplo, bases de datos, canalizaciones de datos y motores de orquestación) directamente en una máquina local o dentro de una canalización de integración continua (CI). Con Docker, puede ejecutar sus pruebas de integración y validación de datos de manera idéntica en todas las plataformas sin contaminar los sistemas operativos locales.
Visual Studio Code (VS Code) es un entorno de desarrollo centralizado para crear secuencias de comandos, depurar, implementar y automatizar pruebas de canalización de datos. Como ingeniero de datos, es posible que lo haya utilizado para sus otros proyectos. Es posible que esté más familiarizado con PyCharm o IntelliJ IDEA. Desde la perspectiva de mi experiencia de usuario, elijo VS Code debido a su construcción liviana, su ecosistema de extensión y su flujo de trabajo híbrido entre notebook y script. Los editores nativos de IA, como Cursor y Windsurf, están ganando rápidamente popularidad entre los desarrolladores, algo que analizaré con más detalle en la última parte de este artículo.
Supongo que ya tienes instalado Python, Poesía y Java. Puede abrir su terminal VS Code, escribir los siguientes scripts para verificar sus versiones y asegurarse de que estén actualizados. También puedes instalarlos debajo de tu terminal si aún no lo has hecho.
python –versión java -versión poesía –versión
La extensión Dev Container le permite utilizar un contenedor Docker como un entorno de desarrollo reproducible y completamente funcional. Estandariza los entornos en todo el equipo y permite probar la lógica de ingesta de datos localmente sin consumir recursos de la nube. Instalar Dev Container es bastante sencillo. Solo necesita abrir Extensiones en VS Code; puede presionar Ctrl+Shift+X (Windows/Linux) o Cmd+Shift+X (Mac), luego buscar ‘Contenedores de desarrollo’ en la barra de búsqueda y hacer clic en “Instalar”.
Pero la extensión Dev Containers no sabe cómo construir su entorno específico. Necesita una ‘guía’. La guía es la carpeta .devcontainer y el archivo devcontainer.json debajo de la carpeta indica la extensión Dev Container:
Qué imagen de Docker descargar. Qué puertos reenviar. Qué extensiones de VS Code instalar dentro del contenedor.
Hay dos métodos para obtener la carpeta .devcontainer. Si es nuevo en estas herramientas, puede utilizar la herramienta automatizada de VS Code. Cuando selecciona una plantilla de Python o Ingeniería de datos, VS Code puede generar la carpeta automáticamente. Cuando tenga más experiencia en este tipo de proyectos, también podrá escribirlo a mano desde cero para cumplir con los requisitos de prueba de su equipo. La carpeta .devcontainer se puede confirmar y enviar a Git, junto con el código fuente y los datos fuente, que se prepara para probar.
Para hacerte la vida más fácil, puedes clonar el repositorio de Git y abrir esa carpeta con VS Code.
clon de git https://github.com/company/data-ingestion-transformation.git
El último paso de la configuración es reabrir el contenedor. ¿Por qué es importante? Porque cuando haces clic en “Reabrir en contenedor”, VS Code reinicia su motor backend. Inicia el contenedor Docker y adjunta la carpeta de su proyecto local directamente dentro de ese contenedor. Su código fuente y sus datos fuente en esta canalización ETL son accesibles desde el entorno Docker. Puede ejecutar sus pruebas de forma segura en una zona de pruebas aislada. ¿Suena bien? Sí, ahora ya configuró su entorno y está listo para comenzar a probar sus canalizaciones ETL.
Deje que las pruebas le digan qué hace el sistema
Cuando heredo una canalización ETL desconocida, mi primera pregunta no es: “¿Cómo funciona el código?” En cambio, pregunto: “¿Qué comportamiento se espera que produzca el sistema?” Las pruebas suelen responder a esa pregunta más rápido que el código fuente.
Imagine que la empresa a la que se une utiliza LLM como GPT-5.5, Claude 4.6 y Gemini 3 Pro y el equipo de Finanzas quiere realizar un seguimiento del gasto en IA en todos los equipos.
La tabla anterior muestra parte de los datos en formato csv que se almacenarán. Los nombres de las columnas deben estandarizarse reemplazando los espacios con guiones bajos para que los sistemas posteriores puedan hacer referencia a los campos de manera consistente. Por ejemplo. ‘Nombre del modelo’ debería convertirse en ‘Nombre_modelo’. Encontró ingest.py para definir las funciones para la estandarización de columnas y la ingesta de datos y ai_cost_ingest.py para llamar a estas funciones en la carpeta.
importar el registro desde escribir importar Lista desde pyspark.sql importar SparkSession def sanitize_columns(columnas: Lista[str]) -> Lista[str]: devolver [column.replace(” “, “_”) for column in columns]
def run(spark: SparkSession, ingest_path: str, transform_path: str) -> Ninguno: logging.info(“Leyendo archivo de texto de: %s”, ingest_path) input_df = ( spark.read.format(“org.apache.spark.csv”) .option(“header”, True) .csv(ingest_path) ) rename_columns = sanitize_columns(input_df.columns) ref_df = input_df.toDF(*renamed_columns) ref_df.write.parquet(transformation_path) importar registro importar sistema desde pyspark.sql importar SparkSession desde data_ingestions.ai_cost importar ingesta LOG_FILENAME = “project.log” APP_NAME = “AI_Cost Pipeline: Ingest” if __name__ == “__main__”: logging.basicConfig(filename=LOG_FILENAME, nivel=logging.INFO) logging.info(sys.argv) if len(sys.argv) != 3: logging.warning(“Se requieren fuente de entrada y ruta de salida”) sys.exit(1) spark = SparkSession.builder.appName(APP_NAME).getOrCreate() sc = spark.sparkContext nombre_aplicación = sc.appName logging.info(“Aplicación inicializada: ” + nombre_aplicación) input_path = sys.argv[1]
ruta_salida = sys.argv[2]
ingest.run(spark, input_path, output_path) logging.info(“Aplicación realizada: ” + spark.sparkContext.appName) spark.stop()
Primero debe comprender las funciones definidas. Quizás se pregunte: “¿Qué debería hacer exactamente sanitize_columns()? ¿Maneja espacios iniciales, espacios finales y espacios internos?” Con estas preguntas en mente, escribe dicho código:
de data_ingestions.ai_cost importar ingesta def test_should_sanitize_nothing() -> Ninguno: no_whitespace_columns = [“Model”]
actual = ingest.sanitize_columns(no_whitespace_columns) esperado = no_whitespace_columns afirmar esperado == real def test_should_sanitize_whitespace_outside() -> Ninguno: no_whitespace_columns = [” Prompt Tokens “]
real = ingest.sanitize_columns(no_whitespace_columns) esperado = [“_Prompt_Tokens_”]
afirmar esperado == definición real test_should_sanitize_whitespace_in_between() -> Ninguno: no_whitespace_columns = [“Prompt Tokens”]
real = ingest.sanitize_columns(no_whitespace_columns) esperado = [“Prompt_Tokens”]
afirmar esperado == real
El código le permite probar la función de sanitize_columns() directamente sin iniciar Spark ni procesar archivos. Es un ejemplo de prueba unitaria.
Pruebas unitarias
Las pruebas unitarias están diseñadas para validar una pequeña parte de la lógica de forma aislada. Suelen ser rápidos, deterministas e independientes de sistemas externos.
Pruebas de integración
Las pruebas unitarias indican si una pequeña parte de la lógica se comporta correctamente. Pero no pueden responder a la pregunta: “¿Funciona toda la tubería cuando todos los componentes están conectados entre sí?”
Para un ingeniero de datos, esto suele significar:
Leer archivos Iniciar Spark Ejecutar transformaciones Escribir resultados Validar resultados
Para probar todo el proceso, necesitamos pruebas de integración, que revelan el comportamiento del sistema. Las pruebas de integración son muy útiles durante la incorporación porque describen lo que debe hacer el sistema, independientemente de cómo evolucione la implementación con el tiempo.
Para el proyecto de ingesta de datos AI_cost, puede utilizar una prueba de integración para ayudar a validar si:
La entrada llega como archivos CSV. Spark se utiliza para procesar los datos. Los nombres de las columnas están desinfectados. Los valores de los datos permanecen sin cambios. La salida está escrita en formato Parquet. El flujo de trabajo de ingesta completo debe realizarse correctamente. importar csv importar sistema operativo importar archivo temporal desde pathlib importar ruta desde escribir importar lista, tupla desde pyspark.sql importar SparkSession desde data_ingestions.ai_cost importar ingesta def test_should_sanitize_column_names( spark_session: SparkSession, ) -> Ninguno: dada_ingest_folder, dada_transform_folder = ( __create_ingest_and_transform_folders() ) input_csv_path = carpeta_ingesta_dada + “input.csv” csv_content = [
[
“Model Name”,
“Prompt Tokens”,
” Completion Tokens ”
],
[
“GPT-5.5”,
“1200”,
“300”
],
[
“Gemini 3 Pro”,
“900”,
“250”
]]__write_csv_file(input_csv_path, csv_content) ingest.run( spark_session, input_csv_path, dada_transform_folder ) real = spark_session.read.parquet( dada_transform_folder ) esperado = spark_session.createDataFrame(
[
[“GPT-5.5”, “1200”, “300”],
[“Gemini 3 Pro”, “900”, “250”]
],
[
“Model_Name”,
“Prompt_Tokens”,
“_Completion_Tokens_”
]
) afirmar esperado.collect() == actual.collect()
Deje que la IA lea el proceso ETL antes que usted
Imagine que está revisando una canalización ETL desconocida que contiene cientos o incluso miles de líneas de código PySpark. Comprender el código y escribir pruebas puede llevar horas o incluso días. Hoy en día, herramientas como Cursor, Windsurf y GitHub Copilot pueden ayudar a acelerar este proceso.
Tome Cursor como ejemplo. Como asistente de IA, puede analizar un repositorio completo y generar explicaciones de módulos, funciones y flujos de datos individuales. También puede generar versiones iniciales de pruebas unitarias y pruebas de integración. Para maximizar su productividad, debe hacer las preguntas correctas como ingeniero de datos. Aquí algunos ejemplos de preguntas que puede hacer:
¿Cuál es el propósito de este trabajo ETL? ¿Qué formatos de entrada y salida espera este canal? ¿Qué funciones son responsables de la validación de datos? ¿Qué casos extremos no se han probado actualmente?
La IA puede sugerir casos de prueba, pero no puede determinar si esas pruebas cumplen con los requisitos comerciales y la estrategia de la empresa. Comprender el proceso, validar las suposiciones y revisar el código sigue siendo su responsabilidad. La IA es un acelerador de la productividad más que un sustituto del criterio de ingeniería. Le ahorra tiempo al comprender y probar la canalización de ETL para que pueda concentrarse en trabajos de ingeniería de datos de mayor valor, como diseñar arquitecturas de datos, crear plataformas de datos escalables y potenciar la toma de decisiones basada en datos.