Skip to content

Instantly share code, notes, and snippets.

@ColinMaudry

ColinMaudry/traitement-du-flux.md

Last active November 11, 2020 14:45
Show Gist options
  • Save ColinMaudry/13f8e16fc2dbf3a6b6e712879b6cb735 to your computer and use it in GitHub Desktop.
Save ColinMaudry/13f8e16fc2dbf3a6b6e712879b6cb735 to your computer and use it in GitHub Desktop.
Download ZIP
Proposition pour le traitement du flux dans decp-rama
Raw
traitement-du-flux.md

Proposition pour le traitement du flux dans decp-rama

Contexte

Actuellement, pour chaque source, decp-rama (code source) :

  1. télécharge l'ensemble des fichiers
  2. traite l'ensemble des fichiers
  3. fusionne les données de la source en un seul fichier (ex : json/data.gouv.fr_pes.json)
  4. fusionne l'ensemble des fichiers json/*.json des différentes sources dans un fichier consolidé json/decp.json.
  5. compare le précédent json/decp.json consolidé avec celui du jour pour extraire les nouveaux marchés du jour (=> opération très longue s'il y a de nombreux nouveaux marchés, autrement la moyenne est à 30 minutes pour 300 marchés)
  6. remplace le précédent json/decp.json consolidé par celui du jour
  7. publie les données du jour (ex : decp-2020-11-04.json et decp-2020-11-04.xml)

L'extraction des nouveaux marchés prend actuellement trop de temps, et cela va empirer avec la croissance des données. Plus d'informations sur l'issue Github.

Cibler les fichiers à traiter

Pour réduire le temps de traitement, il ne faut récupérer depuis les différentes sources que les données qui n'ont pas encore été traitées. Ainsi, le temps consacré aux étapes de correction (fix.sh) et de conversion (convert.sh) sera réduit (il est déjà assez court, en moyenne 15 minutes par jour pour l'ensemble des sources). Surtout, les données du jour seront déjà isolées. L'étape de comparaison avec les données de la veille sera donc supprimée (30 minutes par jour en moyenne).

J'identifie trois moyens potentiels de distinguer les nouvelles données pouvant s'appliquer aux sources de DECP :

  1. s'appuyer sur les identifiants de marchés (.uid) : ne traiter que les marchés dont l'uid n’apparaît pas dans la liste des marchés traités
  2. s'appuyer sur les noms de fichiers : ne traiter que les fichiers dont le nom n’apparaît pas dans la liste des fichiers traités
  3. s'appuyer sur les dates publication (métadonnées data.gouv.fr, en-tête HTTP Last-Modified, champ datePublicationDonnees) : ne traiter que les fichiers publiés à une date postérieure au dernier traitement

S'appuyer sur les identifiants de marchés

Cette approche est actuellement utilisée lorsque le nombre de nouveaux marchés est inférieur à 1 000. Elle a l'avantage d'opérer à une maille qui permet une extraction fine des marchés qui n'ont pas encore traité, l'unicité des uid étant jugée fiable (à l'exception des uid incomplets dépourvus du SIRET de l'acheteur ou d'id de marché).

La liste des uid du ficher consolidé du jour est comparée avec la liste des uid de la veille. Établir cette liste est rapide (quelques secondes) :

  1. On extrait la liste des uid du précédent fichier consolidé
jq -r '.marches[].uid' $oldFile`
  1. On extrait la liste des uid du fichier consolidé du jour (même commande)
  2. On fait un diff des deux
diff -u --suppress-common-lines oldMarchesNoDuplicates newMarchesNoDuplicates | grep -e "^+\\w" | sed -E 's/^\\+//' | sort -u > todayMarches

En revanche extraire les données (objets JSON) de quelques centaines de marchés du fichier consolidé (json/decp.json) à partir de cette liste de nouveaux uid pour produire les données du jour (decp-2020-11-04.json) est très long, car l'ensemble du fichier consolidé (173 Mo) doit être parsé pour chaque uid.

L'opération inverse (pour chaque marché du fichier consolidé, vérifier la présence de l'uid dans la liste des uid des nouveaux marchés) ne serait pas assez rapide compte tenu du nombre de marchés (177 000 au 6 novembre 2020). Il faudrait atteindre la vitesse de 5 667 marchés traités par minute (94 par seconde) pour égaler le temps d'extraction actuel (30 minutes).

Lorsque le nombre de nouveaux uid est supérieur à 1 000, on effectue une soustraction du précédent array de marchés à l'array de marchés du jour :

jq --slurpfile previous $oldFile '{"marches": (.marches - $previous\[0\].marches)} ' $newFile

Bien que lente compte tenu de la taille des arrays à comparer, cette méthode reste plus rapide que la méthode par comparaison des uid ci-dessus lorsque le nombre de nouveaux uid est important.

S'appuyer sur les URL

L'ensemble des sources publie un ou plusieurs fichiers pour publier de nouvelles données et les fichiers publiés ne sont jamais modifiés. De plus, le nom, et donc l'URL de chaque fichier de données, est unique.

decp-rama pourrait donc, à chaque phase de traitement réussie, enregistrer la liste des URL traitées pendant cette phase, pour chaque source. À la phase de traitement suivante (par exemple le lendemain), seuls les URL qui n'apparaissent pas dans la liste des URL déjà traitées seront téléchargées et traitées.

Exemple pour la source data.gouv.fr_pes :

  1. l'URL des fichiers déjà traités est enregistrée dans sources/data.gouv.fr_pes.processed et suivie dans le dépôt Git (18 324 URLs au 10/11/2020) :
https://files.data.gouv.fr/decp/24450046800040/2020/08/14/DECP-24450046800040-2020-08-14-01.xml
https://files.data.gouv.fr/decp/24450046800040/2020/08/25/DECP-24450046800040-2020-08-25-01.xml
https://files.data.gouv.fr/decp/24450046800040/2020/08/27/DECP-24450046800040-2020-08-27-01.xml
[...]
  1. le processus decp-gw, qui publie les données de la source data.gouv.fr_pes sur https://files.data.gouv.fr, pourrait être amélioré pour produire, à chaque run, un fichier contenant l'ensemble des URL publiées. Ceci éviterait à decp-rama d'établir cette liste à distance via des centaines de requêtes HTTP.
  2. lors de la phase téléchargement de data.gouv.fr_pes (script actuel), decp-rama compare la liste des fichiers produite par decp-gw avec sources/data.gouv.fr_pes.processes. Les URL ne figurant pas dans ce fichier sont considérées comme nouvelles, et sont donc téléchargées
  3. le traitement de la source continue comme avant (corrections, conversion vers JSON), mais sur les nouvelles données plutôt que sur l'ensemble des données de la source
  4. les autres sources sont traitées selon une approche similaire (la liste des URL est plus facile à générer pour les autres sources, notamment celles publiées sur data.gouv.fr)
  5. les données produites pour chacune des sources sont fusionnées pour produire les données du jour
  6. les données du jour sont ajoutées au fichier consolidé actuellement publié sur data.gouv.fr (decp.json)
  7. le fichier des données consolidées, enrichi des données du jour, est dédoublonné par .uid (certaines données de marchés peuvent être publiées via plusieurs sources)
  8. le fichier des données consolidées, enrichi et dédoublonné, et les données du jour, sont publiées sur data.gouv.fr

S'appuyer sur les dates et heure de publication

Cette approche pourrait théoriquement être mise en oeuvre en s'appuyant :

  • sur le champ DECP datePublicationDonnees
  • sur l'en-tête HTTP Last-Modified
  • sur la métadonnée last_modified de data.gouv.fr

Le champ datePublicationDonneesest toujours renseigné, mais la date ne correspond pas forcément à la date de mise à disposition de decp-rama. Il peut s'agir d'une date de publication antérieure dans son cycle de vie. Ce n'est donc pas une donnée fiable pour discriminer les marchés à traiter.

Pour les sources publiées sur data.gouv.fr, et plus généralement les sources divisées en fichiers quotidiens, decp-rama pourrait s'appuyer sur l'en-tête HTTP Last-Modified. Sur data.gouv.fr, pour un fichier donné, la métadonnée last_modified a la même valeur que l'en-tête HTTP Last-Modified. Pour les sources actuellement identifiées, cette donnée est considérée comme fiable.

Cependant, cette approche comporte quelques inconvénients et risques :

  • vérifier la date de publication des milliers de fichiers de la source data.gouv.fr_pes serait très chronophage
  • comment déterminer la date et l'heure du traitement de chaque source ? Au début du téléchargement ? À la fin ? Si un fichier est publié pendant le téléchargement, il risque soit d'être traité de nouveau le lendemain, soit d'être ignoré.
  • la comparaison des dates et heure devrait tenir compte des fuseaux horaires, qui peuvent différer entre les serveurs de publication des données et le serveur qui exécute decp-rama. Cette gestion apporterait de la complexité susceptible de fragiliser le script.

Conclusions

L'approche s'appuyant sur les URL des fichiers est privilégiée compte tenu

  • de sa fiabilité,
  • de sa simplicité de mise en place,
  • de sa compatibilité avec les modalités de publication des différentes sources de données,
  • de sa "debuggabilité".

Les développements suivants seraient nécessaires :

  • dgfip-gw :
    • à la fin de chaque run, mettre à jour et publier la liste de l'ensemble des URL des fichiers de données
  • decp-rama :
    • les scripts de téléchargement de toutes les sources (scripts/sources/$source/get.sh) devront être adaptés pour
      • établir une liste des fichiers publiés
      • la comparer avec la liste des fichiers traités (sources/$source.processed)
      • extraire les nouvelles URL
    • le script de rassemblement des données devra être adapté pour alimenter le fichier consolidé, et non le recréer
    • le script de publication devra ajouter, pour chaque source, la liste des fichiers traités et publiés, pousser ces fichiers mis à jour sur le dépôt (à moins que l'on s'appuie sur le cache de CircleCI)

Le traitement de l'ensemble des données (mode actuel) pourrait être conservé.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment