por Mark Tse, Desarrollador de Software, Desarrollo de Plataformas D2L
publicado originalmente el 24 de marzo de 2018
Agregar una columna, siempre que se anexe al final de un conjunto de datos y, por lo tanto, no altere el orden de las columnas existentes, se considera de bajo impacto. Cualquier sistema que ingiera las versiones anteriores del conjunto de datos que no contuviera estas columnas no debería producir errores o no poder funcionar debido a la adición de esas columnas anexadas.
En este artículo, realizaremos cambios en el archivo Ejemplo de código en Ejemplo de cliente sin encabezado de conjuntos de datos de Brightspace y Conjuntos de datos de Brightspace: ejemplo de cliente de conjuntos de datos diferenciales para admitir cambios aditivos en el esquema de conjuntos de datos de Brightspace.
Resumen de cambios
Para admitir extracciones con columnas en el CSV que la tabla de la base de datos no tiene, tendremos que truncar esas columnas adicionales antes de actualizar la tabla de la base de datos.
Ignorar nuevas columnas
Recuerde que para cada extracto de datos, cargamos todo el extracto en una tabla temporal antes de hacer un upsert (insertar o actualizar):
def update_db(db_conn_params, tabla, csv_data): ''' En una sola transacción, actualice la tabla de la siguiente manera: - Cargar los datos CSV en una tabla temporal - Ejecute una consulta de actualización o inserción para actualizar la tabla principal con los datos de la tabla temporal - Elimine la tabla temporal ''' con psycopg2.connect(**db_conn_params) como conn: con conn.cursor() como cur: tmp_table_id = sql. Identifier('tmp_' + tabla) # 1) Crea la tabla temporal cur.execute( sql. SQL(''' CREATE TEMP TABLE {tmp_table} AS SELECT * FROM {table} LIMIT 0; ''') .format( tmp_table=tmp_table_id, tabla=sql. Identificador (tabla) ) ) # 2) Actualización masiva del CSV en la tabla temporal cur.copy_expert( sql. SQL(''' COPY {tmp_table} DE STDIN CON (FORMATO CSV, ENCABEZADO); ''') .format(tmp_table=tmp_table_id), csv_data ) # 3) Realice el upsert upsert_query_file = os.path.join( os.path.dirname(os.path.abspath(__file__)), 'schema', 'upserts', 'upsert_{table}.sql'.format(table=table) ) con open(upsert_query_file) como upsert_query: cur.execute(upsert_query.read()) # 4) Eliminar la tabla temporal cur.execute(sql. SQL('DROP TABLE {tmp_table}').format(tmp_table=tmp_table_id)) conn.commit()
Paso 2) utiliza el método COPIAR en PostgreSQL, que requiere que el archivo CSV de entrada tenga el número exacto de columnas como en la tabla. Si la base de datos tiene definido un esquema más antiguo y se utiliza un extracto de datos con un esquema más reciente (que tiene más columnas que la tabla), PostgreSQL lanzará una excepción.
Para evitar que esto ocurra, procesaremos el archivo CSV para eliminar todas las columnas superfluas. En primer lugar, tendremos que determinar cuántas columnas tiene la tabla en la base de datos:
def get_number_of_columns(db_conn_params, tabla): con psycopg2.connect(**db_conn_params) como conn: con conn.cursor() como cur: cur.execute( sql. SQL(''' SELECT * FROM {tabla} LIMIT 0; ''') .format(tabla=sql. Identifier(table)) ) # cur.description es una lista de tuplas nombradas # Cada tupla nombrada describe una columna en la tabla # La longitud de esa lista es, por lo tanto, el número de columnas en la # tabla. return len(cur.description)
A continuación, agregaremos un método que procesa la extracción de datos dado el número de columnas de la tabla:
def process_csv_stream(csv_input_stream, num_columns_in_table): ''' Ignorar las columnas excesivas en el CSV debido a cambios aditivos / cambios menores de BDS ignorando cualquier columna en el CSV más allá del número de columnas en la tabla ''' csv_rows = [] # Leer cada línea en el CSV, y truncar todas las columnas aún no definidas para # la tabla en la que estamos cargando los datos csv_reader = csv.reader(csv_input_stream, cotizando=csv. QUOTE_MINIMAL) for line in csv_reader: csv_rows.append(line[:num_columns_in_table]) # Reescribe las filas procesadas en un nuevo objeto similar a un archivo CSV (es decir, un flujo #) csv_data = io. StringIO() csv_writer = csv.writer(csv_data, quoting=csv. QUOTE_MINIMAL) csv_writer.writerows(csv_rows) # Rebobina el flujo hasta el principio antes de devolverlo para que el # lector descendente lea desde el principio csv_data.seek(io. SEEK_SET) Devolver csv_data
Problemas de memoria
Nótese que en process_csv_stream() leemos todo el CSV en la memoria. Esto se vuelve inviable con grandes extracciones de datos. Para evitar problemas de falta de memoria, procesaremos el CSV en lotes en lugar de todo el archivo al mismo tiempo. Para ello, añadiremos un nuevo método:
def batch_update_db(db_conn_params, table, csv_file, batch_size=10000): # Elimina la primera fila, que contiene los encabezados csv_file.readline() num_columns = get_number_of_columns(db_conn_params, table) csv_input_stream = io. StringIO() def update_db_with_batch(input_stream): ''' Método auxiliar que forma un cierre para que no tengamos que pasar muchos de los valores utilizados en este método como argumentos ''' # Rebobina el flujo hasta el principio antes de pasarlo en input_stream.seek(io. SEEK_SET) con process_csv_stream(input_stream, num_columns) como csv_data: # Reutilizar este método ya existente update_db(db_conn_params, table, csv_data) input_stream.close() i = 0 for line in csv_file: csv_input_stream.write(line.decode('utf-8')) i += 1 # Por cada 10000 líneas, actualizamos la tabla antes de continuar. # El tamaño del lote es configurable, y el ejemplo de código se puede extender a # pasar el tamaño del lote como argumento (fuera del alcance de este artículo) if i == batch_size: update_db_with_batch(csv_input_stream) # Asignar un nuevo objeto similar a un archivo para que Python sepa que el anterior # se puede limpiar csv_input_stream = io. StringIO() i = 0 # Actualiza la base de datos con las líneas restantes update_db_with_batch(csv_input_stream)
Luego cambiamos nuestro código para usar .batch_update_db() En lugar de update_db() directamente:
def unzip_and_update_db(response_content, db_conn_params, Tabla): con IO. BytesIO(response_content) como response_stream: con zipfile. ZipFile(response_stream) como zipped_data_set: files = zipped_data_set.namelist() assert len(files) == 1 csv_name = files[0] con zipped_data_set.open(csv_name) como csv_file: # update_db(db_conn_params, table, csv_data) batch_update_db(db_conn_params, table, csv_file)
Finalmente, dado que ahora estamos eliminando el encabezado en el código, debemos decirle a PostgreSQL que no elimine la primera fila del CSV eliminando el archivo ENCABEZADO en la opción update_db() método:
cur.copy_expert( sql. SQL(''' COPIAR {tmp_table} DE STDIN CON (FORMATO CSV); ''') .format(tmp_table=tmp_table_id), csv_data)
Opciones alternativas
Estos cambios son necesarios, ya que el ejemplo de código usa PostgreSQL como base de datos de respaldo, que no tiene compatibilidad integrada con SQL para omitir columnas en un archivo CSV. Hay opciones alternativas disponibles para los siguientes motores de base de datos que utilizan la funcionalidad integrada en el propio motor:
Es probable que haya una mejora del rendimiento mediante el uso de la funcionalidad integrada en el motor de base de datos. Sin embargo, los cambios descritos en este artículo son independientes del motor de base de datos y se pueden adaptar a cualquier motor de base de datos. Depende del lector decidir qué opción seguir.
Resumen
En este artículo, hemos mostrado cómo podemos preprocesar un extracto de datos en código para administrar cambios aditivos en el esquema BDS. ¡Háganos saber cualquier comentario, pregunta o sugerencia!