Comparación de esquema entre Parquet y Athena

 Versión 1:

from pyspark.sql import SparkSession

# Crear la sesión de Spark

spark = SparkSession.builder \

    .appName("Comparar Esquema de Parquet y Tabla en Athena") \

    .config("hive.metastore.glue.catalog.id", "glue_catalog_id") \

    .enableHiveSupport() \

    .getOrCreate()

# Nombre de la base de datos y tabla en Athena

database_name = "nombre_base_de_datos"

table_name = "nombre_tabla"

# Ruta del archivo Snappy Parquet

parquet_file_path = "ruta_del_archivo.parquet"

# Leer el esquema del archivo Snappy Parquet

parquet_df = spark.read.format("parquet").load(parquet_file_path)

parquet_schema = parquet_df.schema

# Leer el esquema de la tabla en Athena

athena_df = spark.table(f"{database_name}.{table_name}")

athena_schema = athena_df.schema

# Comparar los esquemas

if parquet_schema == athena_schema:

    print("El esquema del archivo Snappy Parquet y la tabla en Athena son iguales.")

else:

    print("El esquema del archivo Snappy Parquet y la tabla en Athena son diferentes.")

# Cerrar la sesión de Spark

spark.stop()

Una mejora al código anterior para que imprima las diferencias entre el esquema del archivo Snappy Parquet y el esquema de la tabla en Athena:

from pyspark.sql import SparkSession # Crear la sesión de Spark spark = SparkSession.builder \ .appName("Comparar Esquema de Parquet y Tabla en Athena") \ .config("hive.metastore.glue.catalog.id", "glue_catalog_id") \ .enableHiveSupport() \ .getOrCreate() # Nombre de la base de datos y tabla en Athena database_name = "nombre_base_de_datos" table_name = "nombre_tabla" # Ruta del archivo Snappy Parquet parquet_file_path = "ruta_del_archivo.parquet" # Leer el esquema del archivo Snappy Parquet parquet_df = spark.read.format("parquet").load(parquet_file_path) parquet_schema = parquet_df.schema # Leer el esquema de la tabla en Athena athena_df = spark.table(f"{database_name}.{table_name}") athena_schema = athena_df.schema # Obtener las columnas presentes en el esquema del archivo Snappy Parquet pero no en el esquema de la tabla en Athena missing_columns_in_athena = set(parquet_schema.fieldNames()) - set(athena_schema.fieldNames()) # Obtener las columnas presentes en el esquema de la tabla en Athena pero no en el esquema del archivo Snappy Parquet missing_columns_in_parquet = set(athena_schema.fieldNames()) - set(parquet_schema.fieldNames()) if missing_columns_in_athena: print("Columnas presentes en el archivo Snappy Parquet pero no en la tabla de Athena:") for column in missing_columns_in_athena: print(column) if missing_columns_in_parquet: print("Columnas presentes en la tabla de Athena pero no en el archivo Snappy Parquet:") for column in missing_columns_in_parquet: print(column) # Cerrar la sesión de Spark spark.stop()

Una mejora adicional al programa anterior para que también imprima las diferencias en el tipo de columna entre el esquema del archivo Snappy Parquet y el esquema de la tabla en Athena:

from pyspark.sql import SparkSession # Crear la sesión de Spark spark = SparkSession.builder \ .appName("Comparar Esquema de Parquet y Tabla en Athena") \ .config("hive.metastore.glue.catalog.id", "glue_catalog_id") \ .enableHiveSupport() \ .getOrCreate() # Nombre de la base de datos y tabla en Athena database_name = "nombre_base_de_datos" table_name = "nombre_tabla" # Ruta del archivo Snappy Parquet parquet_file_path = "ruta_del_archivo.parquet" # Leer el esquema del archivo Snappy Parquet parquet_df = spark.read.format("parquet").load(parquet_file_path) parquet_schema = parquet_df.schema # Leer el esquema de la tabla en Athena athena_df = spark.table(f"{database_name}.{table_name}") athena_schema = athena_df.schema # Obtener las columnas presentes en el esquema del archivo Snappy Parquet pero no en el esquema de la tabla en Athena missing_columns_in_athena = set(parquet_schema.fieldNames()) - set(athena_schema.fieldNames()) # Obtener las columnas presentes en el esquema de la tabla en Athena pero no en el esquema del archivo Snappy Parquet missing_columns_in_parquet = set(athena_schema.fieldNames()) - set(parquet_schema.fieldNames()) # Obtener las columnas con tipos diferentes entre el esquema del archivo Snappy Parquet y el esquema de la tabla en Athena different_type_columns = [] for field in parquet_schema.fields: if field.name in athena_schema: athena_field = athena_schema[field.name] if field.dataType != athena_field.dataType: different_type_columns.append((field.name, field.dataType, athena_field.dataType)) if missing_columns_in_athena: print("Columnas presentes en el archivo Snappy Parquet pero no en la tabla de Athena:") for column in missing_columns_in_athena: print(column) if missing_columns_in_parquet: print("Columnas presentes en la tabla de Athena pero no en el archivo Snappy Parquet:") for column in missing_columns_in_parquet: print(column) if different_type_columns: print("Columnas con tipos diferentes entre el archivo Snappy Parquet y la tabla de Athena:") for column, parquet_type, athena_type in different_type_columns: print(f"Columna: {column}, Tipo en Parquet: {parquet_type}, Tipo en Athena: {athena_type}") # Cerrar la sesión de Spark spark.stop()