
Pour quelqu'un qui recherche un programme simple et rapide qui prend un csv, appelle l'API de validation des destinataires et produit un CSV, ce programme est fait pour vous.
Lors de la création d'applications de messagerie électronique, les développeurs ont souvent besoin d'intégrer plusieurs services et API. Comprendre les fondamentaux de l'API d'email dans l'infrastructure cloud fournit les bases pour développer des outils robustes tels que le système de validation en masse que nous allons créer dans ce guide.
Une des questions que nous recevons occasionnellement est : comment puis-je valider en masse des listes de courriers électroniques avec la validation de destinataire? Il existe deux options ici, une consiste à télécharger un fichier via l'interface utilisateur de SparkPost pour validation, et l'autre consiste à effectuer des appels individuels par courriel à l'API (car l'API est une validation de courriel unique).
La première option fonctionne très bien mais a une limitation de 20Mb (environ 500,000 adresses). Que faire si quelqu'un a une liste de courriers électroniques contenant des millions d'adresses ? Cela pourrait signifier diviser cela en milliers de téléchargements de fichiers CSV.
Étant donné que le téléchargement de milliers de fichiers CSV semble un peu farfelu, j'ai pris ce cas d'utilisation et j'ai commencé à me demander à quelle vitesse je pourrais faire fonctionner l'API. Dans cet article de blog, j'expliquerai ce que j'ai essayé et comment j'en suis finalement arrivé à un programme qui pourrait effectuer environ 100,000 validations en 55 secondes (alors que dans l'interface utilisateur, j'atteignais environ 100,000 validations en 1 minute 10 secondes). Et bien que cela prenne encore environ 100 heures pour effectuer environ 654 millions de validations, ce script peut fonctionner en arrière-plan, économisant ainsi un temps considérable.
La version finale de ce programme peut être trouvée ici.
My first mistake: utiliser Python
Python est l'un de mes langages de programmation préférés. Il excelle dans de nombreux domaines et est incroyablement simple. Cependant, un domaine où il n'excelle pas est celui des processus concurrents. Bien que Python ait la capacité d'exécuter des fonctions asynchrones, il a ce que l'on appelle le verrou global de l'interpréteur Python ou GIL.
« Le verrou global de l'interpréteur Python ou GIL, en termes simples, est un mutex (ou un verrou) qui permet à un seul thread de garder le contrôle de l'interpréteur Python.
Cela signifie qu'un seul thread peut être en état d'exécution à tout moment. L'impact du GIL n'est pas visible pour les développeurs qui exécutent des programmes mono-thread, mais il peut être un goulet d'étranglement de performances dans le code lié au CPU et multithread.
Comme le GIL ne permet qu'à un seul thread de s'exécuter à la fois, même dans une architecture multithread avec plus d'un cœur de CPU, le GIL s'est acquis la réputation de fonctionnalité « tristement célèbre » de Python. » (https://realpython.com/python-gil/) »
Au départ, je n'étais pas au courant du GIL, alors j'ai commencé à programmer en Python. À la fin, même si mon programme était asynchrone, il était bloqué, et peu importe combien de threads j'ajoutais, je n’obtenais toujours qu'environ 12-15 itérations par seconde.
La partie principale de la fonction asynchrone en Python peut être vue ci-dessous :
async def validateRecipients(f, fh, apiKey, snooze, count): h = {'Authorization': apiKey, 'Accept': 'application/json'} with tqdm(total=count) as pbar: async with aiohttp.ClientSession() as session: for address in f: for i in address: thisReq = requests.compat.urljoin(url, i) async with session.get(thisReq, headers=h, ssl=False) as resp: content = await resp.json() row = content['results'] row['email'] = i fh.writerow(row) pbar.update(1)
Donc, j'ai abandonné l'utilisation de Python et suis retourné à la planche à dessin…
Je me suis décidé à utiliser NodeJS en raison de sa capacité à effectuer des opérations d'entrée/sortie non bloquantes très efficacement. Une autre excellente option pour gérer le traitement asynchrone des API est de construire des consommateurs de webhook sans serveur avec Azure Functions, qui peuvent gérer efficacement des charges de travail variables. Je suis également assez familier avec la programmation en NodeJS.
En utilisant les aspects asynchrones de NodeJS, cela a fini par bien fonctionner. Pour plus de détails sur la programmation asynchrone dans NodeJS, voir https://blog.risingstack.com/node-hero-async-programming-in-node-js/
Ma deuxième erreur : essayer de lire le fichier en mémoire
Décomposition du code final
Après avoir lu et validé les arguments du terminal, j'exécute le code suivant. Tout d'abord, je lis le fichier CSV des emails et compte chaque ligne. Il y a deux objectifs à cette fonction, 1) elle me permet de rendre compte précisément de la progression du fichier [comme nous le verrons plus tard], et 2) elle me permet d'arrêter un chronomètre lorsque le nombre d'emails dans le fichier est égal aux validations terminées. J'ai ajouté un chronomètre pour pouvoir exécuter des benchmarks et m'assurer d'obtenir de bons résultats.
let count = 0; //Line count require("fs") .createReadStream(myArgs[1]) .on("data", function (chunk) { for (let i = 0; i < chunk.length; ++i) if (chunk[i] == 10) count++; }) //Lit le fichier d'entrée et augmente le compteur pour chaque ligne .on("close", function () { //À la fin du fichier d'entrée, après que toutes les lignes aient été comptées, exécutez la fonction de validation des destinataires validateRecipients.validateRecipients(count, myArgs); });
J'appelle ensuite la fonction validateRecipients. Notez que cette fonction est asynchrone. Après avoir vérifié que le fichier d'entrée et le fichier de sortie sont au format CSV, j'écris une ligne d'en-tête et démarre un chronomètre de programme en utilisant la bibliothèque JSDOM.
async function validateRecipients(email_count, myArgs) { if ( //Si les deux fichiers d'entrée et de sortie sont au format .csv extname(myArgs[1]).toLowerCase() == ".csv" && extname(myArgs[3]).toLowerCase() == ".csv" ) { let completed = 0; //Compteur pour chaque appel API email_count++; //Le compteur de lignes renvoie #lignes - 1, cela est fait pour corriger le nombre de lignes //Démarrer un chronomètre const { window } = new JSDOM(); const start = window.performance.now(); const output = fs.createWriteStream(myArgs[3]); //Fichier de sortie output.write( "Email,Valid,Result,Reason,Is_Role,Is_Disposable,Is_Free,Delivery_Confidence\n" ); //Écrire les en-têtes dans le fichier de sortie
Le script suivant est vraiment l'essentiel du programme donc je vais le décomposer et expliquer ce qui se passe. Pour chaque ligne du fichier d'entrée :
Prendre asynchronement cette ligne et appeler l'API de validation des destinataires.
fs.createReadStream(myArgs[1]) .pipe(csv.parse({ headers: false })) .on("data", async (email) => { let url = SPARKPOST_HOST + "/api/v1/recipient-validation/single/" + email; await axios .get(url, { headers: { Authorization: SPARKPOST_API_KEY, }, }) //Pour chaque ligne lue dans le fichier d'entrée, appeler l'API SparkPost Recipient Validation
Ensuite, sur la réponse
Ajouter l'email au JSON (pour pouvoir imprimer l'email dans le CSV)
Valider si la raison est nulle, et si c'est le cas, remplir une valeur vide (c'est pour que le format CSV soit cohérent, car dans certains cas la raison est donnée dans la réponse)
Définir les options et les clés pour le module json2csv.
Convertir le JSON en CSV et le sortir (en utilisant json2csv)
Écrire les progrès dans le terminal
Enfin, si le nombre d'emails dans le fichier = validations complétées, arrêter le chronomètre et imprimer les résultats
.then(function (response) { response.data.results.email = String(email); //Ajouter l'email comme paire clé/valeur à la réponse JSON pour être utilisé pour la sortie response.data.results.reason ? null : (response.data.results.reason = ""); //Si la raison est nulle, la définir sur vide pour que le CSV soit uniforme //Utilise json-2-csv pour convertir le JSON en format CSV et le sortir let options = { prependHeader: false, //Désactive les valeurs JSON d'être ajoutées comme lignes d'en-tête pour chaque ligne keys: [ "results.email", "results.valid", "results.result", "results.reason", "results.is_role", "results.is_disposable", "results.is_free", "results.delivery_confidence", ], //Définit l'ordre des clés }; let json2csvCallback = function (err, csv) { if (err) throw err; output.write(`${csv}\n`); }; converter.json2csv(response.data, json2csvCallback, options); completed++; //Augmenter le compteur API process.stdout.write(`Terminé avec ${completed} / ${email_count}\r`); //Sortie de l'état de Terminé / Total sur la console sans afficher de nouvelles lignes //Si tous les emails ont été validés if (completed == email_count) { const stop = window.performance.now(); //Arrêter le chronomètre console.log( `Tous les emails ont été validés avec succès en ${ (stop - start) / 1000 } secondes` ); } })
Un problème final que j'ai trouvé était que cela fonctionnait très bien sur Mac, mais que j'ai rencontré l'erreur suivante en utilisant Windows après environ 10 000 validations :
Erreur : connect ENOBUFS XX.XX.XXX.XXX:443 – Local (non défini:non défini) avec l'email XXXXXXX@XXXXXXXXXX.XXX
Après avoir fait des recherches supplémentaires, il semble s'agir d'un problème avec le pool de connexions du client HTTP NodeJS qui ne réutilise pas les connexions. J'ai trouvé cet article Stackoverflow sur le problème, et après avoir creusé davantage, j'ai trouvé une bonne configuration par défaut pour la bibliothèque axios qui a résolu ce problème. Je ne suis toujours pas certain pourquoi ce problème ne survient que sur Windows et pas sur Mac.
Étapes suivantes
Pour quelqu'un qui cherche un programme simple et rapide qui prend un fichier CSV, appelle l'API de validation des destinataires et produit un fichier CSV, ce programme est pour vous.
Quelques ajouts à ce programme seraient les suivants :
Construire un front-end ou une interface utilisateur plus simple à utiliser
Meilleure gestion des erreurs et des tentatives de reprise car si pour une raison quelconque l'API génère une erreur, le programme ne réessaie pas actuellement l'appel
Envisager la mise en œuvre en tant que Fonction Azure sans serveur pour un ajustement automatique et une gestion réduite de l'infrastructure
Je serais également curieux de voir si des résultats plus rapides pourraient être obtenus avec un autre langage tel que Golang ou Erlang/Elixir. Au-delà du choix du langage, les limitations de l'infrastructure peuvent également avoir un impact sur les performances - nous avons appris cela de première main lorsque nous avons atteint des limites DNS non documentées dans AWS qui ont affecté nos systèmes de traitement de courriels à haut volume.
Pour les développeurs intéressés par la combinaison du traitement API avec des outils de flux de travail visuels, découvrez comment intégrer Flow Builder avec Google Cloud Functions pour des flux de travail d'automatisation sans code.
N'hésitez pas à me fournir tout retour ou suggestion pour l'expansion de ce projet.