par Mark Tse, développeur de logiciels, Développement de plate-forme D2L
publié initialement le 24 mars 2018
L’ajout d’une colonne, à condition qu’elle soit ajoutée à la fin d’un ensemble de données et ne perturbe donc pas l’ordre des colonnes existantes, est considéré comme ayant un faible impact. Tout système ingérant les versions précédentes de l’ensemble de données qui ne contenaient pas ces colonnes ne doit pas lancer d’erreurs ni être incapable de fonctionner en raison de l’ajout de ces colonnes ajoutées.
Dans cet article, nous apporterons des modifications à la exemple de code dans Exemple de client sans tête Brightspace Data Sets et Brightspace Data Sets - Exemple de client d’ensembles de données différentiels pour prendre en charge les modifications additives apportées au schéma brightspace data sets.
Vue d’ensemble des modifications
Pour prendre en charge les extraits avec des colonnes dans le CSV que la table de base de données n’a pas, nous devrons tronquer ces colonnes supplémentaires avant de mettre à jour la table de base de données.
Ignorer les nouvelles colonnes
Rappelons que pour chaque extrait de données, nous téléverserons l’extrait entier dans une table temporaire avant de faire un upsert (insérer ou mettre à jour) :
def update_db(db_conn_params, table, csv_data) : ''' Dans une seule transaction, mettez à jour la table en : - Chargement des données CSV dans une table temporaire - Exécutez une mise à jour ou insérez une requête pour mettre à jour la table principale avec les données dans la table temporaire - Supprimer la table temporaire ''' avec psycopg2.connect(**db_conn_params) comme conn : avec conn.cursor() comme cur : tmp_table_id = sql. Identifier('tmp_' + table) # 1) Créez la table temporaire cur.execute( sql. SQL(''' CREATE TEMP TABLE {tmp_table} AS SELECT * FROM {table} LIMIT 0 ; ''') .format( tmp_table=tmp_table_id, table=sql. Identificateur(table) ) ) # 2) Mise à jour en bloc du CSV dans la table temporaire cur.copy_expert( sql. SQL(''' COPY {tmp_table} FROM STDIN WITH (FORMAT CSV, HEADER) ; ''') .format(tmp_table=tmp_table_id), csv_data ) # 3) Effectuez l’upsert upsert_query_file = os.path.join( os.path.dirname(os.path.abspath(__file__)), 'schema', 'upserts', 'upsert_{table}.sql'.format(table=table) ) avec open(upsert_query_file) comme upsert_query : cur.execute(upsert_query.read()) # 4) Supprimer la table temporaire cur.execute(sql. SQL('DROP TABLE {tmp_table}').format(tmp_table=tmp_table_id)) conn.commit()
Étape 2) utilise le COPY dans PostgreSQL, qui nécessite que le fichier CSV d’entrée ait le nombre exact de colonnes comme dans la table. Si un schéma plus ancien est défini dans la base de données et qu’un extrait de données avec un schéma plus récent est utilisé (qui contient plus de colonnes que la table), PostgreSQL lève une exception.
Pour éviter que cela ne se produise, nous traiterons le fichier CSV pour supprimer toutes les colonnes superflues. Tout d’abord, nous devrons déterminer le nombre de colonnes de la table dans la base de données :
def get_number_of_columns(db_conn_params, table) : avec psycopg2.connect(**db_conn_params) comme conn : avec conn.cursor() comme cur : cur.execute( sql. SQL(''' SELECT * FROM {table} LIMIT 0 ; ''') .format(table=sql. Identifier(table)) ) # cur.description est une liste de tuples nommés # Chaque tuple nommé décrit une colonne dans la table # La longueur de cette liste est donc le nombre de colonnes dans la table #. return len(cur.description)
Ensuite, nous allons ajouter une méthode qui traite l’extrait de données compte tenu du nombre de colonnes dans la table :
def process_csv_stream(csv_input_stream, num_columns_in_table) : ''' Ignorez les colonnes excessives dans le CSV en raison de changements additifs / modifications mineures BDS en ignorant toutes les colonnes du CSV au-delà du nombre de colonnes dans le tableau ''' csv_rows = [] # Lisez chaque ligne dans le CSV, et tronquez toutes les colonnes non encore définies pour # la table dans laquelle nous chargeons les données csv_reader = csv.reader(csv_input_stream, quoting=csv. QUOTE_MINIMAL) pour la ligne dans csv_reader : csv_rows.append(line[ :num_columns_in_table]) # Réécrivez les lignes traitées dans un nouvel objet de type fichier CSV (c’est-à-dire un flux #) csv_data = io. StringIO() csv_writer = csv.writer(csv_data, quoting=csv. QUOTE_MINIMAL) csv_writer.writerows(csv_rows) # Rembobinez le flux au début avant de le retourner afin que le lecteur # en aval lise depuis le début csv_data.seek(io. SEEK_SET) retour csv_data
Problèmes de mémoire
Notez que dans process_csv_stream() nous lisons l’ensemble du CSV en mémoire. Cela devient impossible avec de grands extraits de données. Pour éviter les problèmes de mémoire insuffisante, nous traiterons le fichier CSV par lots au lieu du fichier entier en même temps. Pour ce faire, nous allons ajouter une nouvelle méthode :
def batch_update_db(db_conn_params, table, csv_file, batch_size=10000) : # Supprimez la première ligne, qui contient les en-têtes csv_file.readline() num_columns = get_number_of_columns(db_conn_params, table) csv_input_stream = io. StringIO() déf update_db_with_batch(input_stream) : ''' Méthode d’aide qui forme une fermeture afin que nous n’ayons pas à passer beaucoup de valeurs utilisées dans cette méthode comme arguments ''' # Rembobinez le flux au début avant de le transmettre sur input_stream.seek(io. SEEK_SET) avec process_csv_stream(input_stream, num_columns) comme csv_data : # Réutilisez cette méthode déjà existante update_db(db_conn_params, table, csv_data) input_stream.close() i = 0 pour la ligne dans csv_file : csv_input_stream.write(line.decode('utf-8')) i += 1 # Pour chaque 10000 lignes, nous mettons à jour le tableau avant de continuer. # Batch size is configurable, and the code sample can be extended to # pass the batch size in as an argument (out-of-scope of this article) if i == batch_size : update_db_with_batch(csv_input_stream) # Assignez un nouvel objet de type fichier afin que Python sache que l’ancien # peut être nettoyé csv_input_stream = io. StringIO() i = 0 # Mettre à jour la base de données avec les lignes restantes update_db_with_batch(csv_input_stream)
Ensuite, nous changeons notre code pour l’utiliser .batch_update_db() À la place de update_db() directement :
def unzip_and_update_db(response_content, db_conn_params, table) : avec io. BytesIO(response_content) comme response_stream : avec zipfile. ZipFile(response_stream) as zipped_data_set : files = zipped_data_set.namelist() assert len(files) == 1 csv_name = files[0] with zipped_data_set.open(csv_name) as csv_file : # update_db(db_conn_params, table, csv_data) batch_update_db(db_conn_params, table, csv_file)
Enfin, puisque nous supprimons maintenant l’en-tête dans le code, nous devons dire à PostgreSQL de ne pas supprimer la première ligne du CSV en supprimant le HEADER dans l' update_db() méthode :
cur.copy_expert( sql. SQL(''' COPY {tmp_table} FROM STDIN WITH (FORMAT CSV) ; ''') .format(tmp_table=tmp_table_id), csv_data)
Options alternatives
Ces modifications sont requises car l’exemple de code utilise PostgreSQL comme base de données de sauvegarde, qui n’a pas de prise en charge SQL intégrée pour sauter des colonnes dans un fichier CSV. D’autres options sont disponibles pour les moteurs de base de données suivants qui utilisent les fonctionnalités intégrées au moteur lui-même :
Il y a probablement une amélioration des performances en utilisant la fonctionnalité intégrée dans le moteur de base de données. Cependant, les modifications décrites dans cet article sont indépendantes du moteur de base de données et peuvent être adaptées à n’importe quel moteur de base de données. C’est au lecteur de décider quelle option poursuivre.
Résumé
Dans cet article, nous avons montré comment nous pouvons pré-traiter un extrait de données dans le code pour gérer les modifications additives apportées au schéma BDS. S’il vous plaît, faites-nous savoir vos commentaires, questions ou retours !