Nous allons voir ces notions en pratique dans la suite de l’article.<\/p>\n
Pour lancer Kestra en local, le plus simple est d’utiliser le fichier Docker compose fournit<\/a>, puis de le lancer via Ce Docker compose va lancer Kestra en mode standalone<\/strong> (tous les composants en un seul processus) avec une base de donn\u00e9es PostgreSQL comme queue et repository (pour le stockage et l’ex\u00e9cution des flows) et un storage local (pour le stockage des donn\u00e9es des flows).<\/p>\n Kestra est compos\u00e9 de plusieurs composants qui communiquent ensemble via des queues et stockent les informations des flows via des repositories, ces queues et repositories peuvent \u00eatre impl\u00e9ment\u00e9es de diff\u00e9rentes mani\u00e8res (memory, JDBC, Kafka, Elasticsearch). Je ne vais pas entrer en d\u00e9tail dans l’architecture de Kestra aujourd’hui, mais pour les plus curieux vous pouvez vous r\u00e9f\u00e9rer \u00e0 la documentation d’architecture<\/a> de Kestra.<\/p>\n Une fois Kestra d\u00e9marr\u00e9, une interface graphique sera disponible sur le port 8080 : http:\/\/localhost:8080<\/a><\/a>, c’est via cette interface que nous allons faire tous les exemples de cet article.<\/p>\n Au premier lancement, un Guided Tour vous sera propos\u00e9, vous pouvez le suivre ou le passer pour pouvoir suivre les exemples de cet article.<\/p>\n Pour cr\u00e9er un flow, allez dans le menu Flows<\/strong> puis cliquez sur le bouton Create<\/strong> en bas \u00e0 droite. Vous voici maintenant avec un textarea dans lequel vous allez pouvoir entrer le descriptif YAML du flow.<\/p>\n Un flow a :<\/p>\n Ici, nous avons ajout\u00e9 une seule t\u00e2che qui a trois propri\u00e9t\u00e9s :<\/p>\n Chaque t\u00e2che aura ses propres propri\u00e9t\u00e9s, elles sont document\u00e9es dans la documentation en ligne ainsi que dans la documentation int\u00e9gr\u00e9e \u00e0 l’interface graphique de Kestra (Documentation -> Plugins -> puis s\u00e9lectionnez un plugin dans le menu de droite).<\/p>\n Au sein de l’\u00e9diteur, la description YAML du flow est valid\u00e9 et de l’autocompl\u00e9tion est disponible pour le type et les propri\u00e9t\u00e9s des t\u00e2ches via CTRL+SPACE.<\/p>\n Kestra se base sur un syst\u00e8me de plugins qui permet d’ajouter des t\u00e2ches (mais aussi des triggers ou des conditions qu’on verra plus tard). Par d\u00e9faut, il y a tr\u00e8s peu de plugins fournit dans Kestra mais l’image Docker utilis\u00e9e dans le Docker compose est construite en ajoutant tous les plugins maintenus par l’\u00e9quipe de d\u00e9veloppement de Kestra, leur documentation est ici : https:\/\/kestra.io\/plugins\/<\/a><\/a>.<\/p>\n Pour lancer ce premier flow, allez dans l’onglet Executions<\/strong> puis cliquez sur le bouton New Execution<\/strong>.\nVous basculerez alors sur la vue Gantt de l’ex\u00e9cution du flow qui est mise \u00e0 jour en temps r\u00e9elle avec le statut de l’ex\u00e9cution.<\/p>\n L’onglet Logs<\/strong> permet de voir les logs de l’ex\u00e9cution.<\/p>\n On peut remarquer le log Ajoutons maintenant un input nomm\u00e9 name<\/strong> \u00e0 notre flow pour lui donne le nom de la personne \u00e0 qui dire bonjour.\nPour cela, il faut \u00e9diter le flow en s\u00e9lectionnant le flow, soit via la navigation breadcrumb en haut ( Home \/ Flows \/ fr.loicmathieu.example.hello-world), soit en cliquant \u00e0 gauche sur Flows<\/strong> et en le s\u00e9lectionnant dans la liste ; puis aller dans l’onglet Source<\/strong>.<\/p>\n On peut passer des donn\u00e9es \u00e0 un flow via des inputs. Nous allons d\u00e9finir un input name<\/strong> de type STRING<\/strong>.<\/p>\n Quand on clique sur le boutton New Execution pour ex\u00e9cuter un flow, un formulaire nous demande de renseigner les inputs du flow.<\/p>\n Apr\u00e8s avoir rentr\u00e9 une valeur pour le nom et ex\u00e9cut\u00e9 le flow, on peut constater que le log contient bien la valeur de l’input.<\/p>\n Dans cet exemple, le flow va prendre en entr\u00e9e (en input) un fichier CSV, le charger dans Google BigQuery, puis ex\u00e9cuter une requ\u00eate sur les donn\u00e9es charg\u00e9es et afficher dans les logs le r\u00e9sultat.<\/p>\n J’utilise ici BigQuery car c’est plus simple que de d\u00e9marrer une base de donn\u00e9es, mais cela n\u00e9cessite un compte Google Cloud et un service account de configur\u00e9. Pour faire cela, il faut cr\u00e9er un service account<\/a> Google Cloud et le renseigner dans la propri\u00e9t\u00e9 Sans plus tarder, la solution ; )<\/p>\n J’ai d\u00e9fini ici un input de type La t\u00e2che load-beers<\/strong> de type La t\u00e2che query-beers<\/strong> de type La t\u00e2che end-log<\/strong> va \u00e9crire un log, nous l’avons d\u00e9j\u00e0 vu pr\u00e9c\u00e9demment. Ici, nous voulons \u00e9crire dans les logs le nombre d’enregistrements charg\u00e9s depuis la base de donn\u00e9es, nous allons donc r\u00e9cup\u00e9rer l’output correspondant de la t\u00e2che query-beers<\/strong> via une expression Pebble : L’expression Dans cet exemple, nous allons requ\u00eater les 10 pages Wikipedia les plus vues gr\u00e2ce au dataset public BigQuery wikipedia.pageviews_2023<\/strong> pour les langues fran\u00e7ais, anglais et allemand. Puis nous allons transformer le r\u00e9sultat en CSV.<\/p>\n La t\u00e2che for-each-countries<\/strong> de type La t\u00e2che query-top-ten<\/strong> de type La t\u00e2che write-csv<\/strong> de type Apr\u00e8s ex\u00e9cution du flow, vous pouvez aller dans l’onglet Outputs<\/strong> pour acc\u00e9der aux outputs des t\u00e2ches et, entre autres, t\u00e9l\u00e9charger les fichiers CSV g\u00e9n\u00e9r\u00e9s par le flow.<\/p>\n Par d\u00e9faut, un flow ne peut \u00eatre lanc\u00e9 que manuellement via l’interface graphique ou l’API de Kestra.<\/p>\n Il est possible de d\u00e9clancher un flow depuis un \u00e9v\u00e8nement ext\u00e9rieur, c’est le r\u00f4le du trigger.<\/p>\n Kestra inclu de base trois triggers: flow<\/strong>, qui permet de d\u00e9clancher un flow depuis un autre flow, webhook<\/strong> qui permet de d\u00e9clancher un flow depuis une URL webhook et schedule<\/strong> qui permet de d\u00e9clencher un flow p\u00e9riodiquement depuis une expression cron. De nombreux autres triggers sont disponibles au sein des plugins de Kestra et permettent de d\u00e9clancher un flow depuis un message dans un broker, un fichier, ou la pr\u00e9sence d’un enregistrement dans une table de base de donn\u00e9es par exemple.<\/p>\n L’exemple suivant va d\u00e9clencher un flow toutes les minutes, il utilise une cron expression pour d\u00e9finir sa p\u00e9riodicit\u00e9 de d\u00e9clenchement.<\/p>\n Kestra offre des fonctionnalit\u00e9s avanc\u00e9es via des t\u00e2ches permettant d’ex\u00e9cuter des scripts Bash, Python ou NodeJS.<\/p>\n L’exemple suivant va requ\u00eater le m\u00eame dataset BigQuery des pages vues Wikipedia, l’\u00e9crire au format CSV, puis utiliser ce CSV dans une t\u00e2che de type Cette t\u00e2che prend en propri\u00e9t\u00e9s :<\/p>\n Le script Python va utiliser Pandas pour lire le fichier CSV et le transformer en data frame Pandas, puis r\u00e9aliser la somme de la colonne views<\/strong>. Cette somme sera ensuite mise en output de la t\u00e2che gr\u00e2ce \u00e0 la librairie Python Kestra.<\/p>\n Voici les logs d’ex\u00e9cution de ce flow.<\/p>\n Dans cet article introductif, nous avons vu les principaux concepts de Kestra et quelques exemples de flow.<\/p>\n Pour aller plus loin, vous pouvez consulter la documentation en ligne<\/a> ainsi que la liste des plugins<\/a>.<\/p>\n Ketra est un projet communautaire open source disponible sur GitHub, n’h\u00e9sitez pas \u00e0 :<\/p>\n <\/p>","protected":false},"excerpt":{"rendered":" Kestra est un orchestrateur et scheduler de donn\u00e9e open source. Avec Kestra, les workflows de donn\u00e9es, appel\u00e9s flows, utilisent le format YAML et sont ex\u00e9cut\u00e9s par son moteur via un appel API, l’interface utilisateur, ou un trigger (webhook, schedule, SQL query, Pub\/Sub message, \u2026). Les notions importantes de Kestra sont : Le flow : qui d\u00e9crit la mani\u00e8re dont la donn\u00e9e sera orchestr\u00e9e (le workflow donc). C’est une suite de tasks. La task : une \u00e9tape du flow qui va…docker compose up<\/code>.<\/p>\nMon premier flow<\/h2>\n
\nid: hello-world\nnamespace: fr.loicmathieu.example\n\ntasks:\n - id: hello\n type: io.kestra.core.tasks.debugs.Echo\n format: \"Hello World\"\n<\/pre>\n
id<\/code> qui est son identifiant unique au sein d’un namespace.<\/li>\n\nnamespace<\/code>, un namespace est un espace de nommage pour les flux, les namespaces sont hierarchique comme les r\u00e9pertoires d’un syst\u00e8me de fichier.<\/li>\n\ntasks<\/code> qui est la liste des t\u00e2ches \u00e0 r\u00e9aliser \u00e0 l’ex\u00e9cution du flow.<\/li>\n<\/ul>\nid<\/code> qui est son identifiant unique au sein du flow.<\/li>\n\ntype<\/code> qui est le nom de la classe qui repr\u00e9sente la t\u00e2che.<\/li>\n\nformat<\/code>, c’est une propri\u00e9t\u00e9 propre aux t\u00e2ches de type Echo qui d\u00e9finit le format du message \u00e0 logguer (une t\u00e2che de type Echo est comme la commande shell echo).<\/li>\n<\/ul>\n![\"\"](\"https:\/\/i0.wp.com\/loicmathieu.fr\/wordpress\/wp-content\/uploads\/kestra-01.png?resize=640%2C167&ssl=1\")
\n![\"\"](\"https:\/\/i0.wp.com\/loicmathieu.fr\/wordpress\/wp-content\/uploads\/kestra-02.png?resize=640%2C169&ssl=1\")
\nHello World<\/code> g\u00e9n\u00e9r\u00e9 par la t\u00e2che hello<\/strong>.<\/p>\n\nid: hello-world\nnamespace: fr.loicmathieu.example\n\ninputs:\n - type: STRING\n name: name\n\ntasks:\n - id: hello\n type: io.kestra.core.tasks.debugs.Echo\n format: \"Hello {{inputs.name}}\"\n<\/pre>\n![\"\"](\"https:\/\/i0.wp.com\/loicmathieu.fr\/wordpress\/wp-content\/uploads\/kestra-03.png?resize=640%2C164&ssl=1\")
\nHello {{inputs.name}}<\/code> est une expression qui utilise le moteur de templating Pebble qui va faire un rendu de ce qui est entre les moustaches {{<\/code> et }}<\/code>. Ici l’expression Pebble va chercher la valeur de la variable inputs.name<\/strong> qui pointe vers l’input de nom name<\/strong>.<\/p>\nInputs et BigQuery<\/h2>\n
serviceAccount<\/code> de la t\u00e2che BigQuery. Pour \u00e9viter de coder en dure cette variable, il est possible d’utiliser une variable d’environement ou une variable globale au cluster, voir la documentation en ligne sur les variables<\/a>.<\/p>\n\nid: beer\nnamespace: fr.loicmathieu.example\ndescription: A flow to handle my beers data\n\ninputs:\n - type: FILE\n description: Beers data\n name: beers\n\ntasks:\n - id: start-log\n type: io.kestra.core.tasks.debugs.Echo\n format: \"{{taskrun.startDate}} - {{task.id}} - Launching the flow\"\n - id: load-beers\n type: io.kestra.plugin.gcp.bigquery.Load\n serviceAccount: \"\"\n csvOptions:\n fieldDelimiter: \",\"\n destinationTable: beers.beers\n format: CSV\n from: \"{{inputs.beers}}\"\n - id: query-beers\n type: io.kestra.plugin.gcp.bigquery.Query\n serviceAccount: \"\"\n fetchOne: true\n sql: |\n SELECT count(*) as count FROM beers.beers\n - id: end-log\n type: io.kestra.core.tasks.debugs.Echo\n format: \"{{taskrun.startDate}} - {{outputs['query-beers'].row.count}} beers loaded\"\n<\/pre>\nFILE<\/code> nomm\u00e9 beers<\/strong>. Kestra va d\u00e9tecter le type de l’input et g\u00e9n\u00e9rer le formulaire d’ex\u00e9cution du flow correspondant. Cela permettra d’uploader un fichier CSV contenant une liste de bi\u00e8re. Ce fichier sera stock\u00e9 dans le stockage interne de Kestra et disponible dans toutes les t\u00e2ches.<\/p>\nio.kestra.plugin.gcp.bigquery.Load<\/code> permet de charger un fichier CSV dans une table BigQuery. Le dataset BigQuery beers<\/strong> doit avoir \u00e9t\u00e9 cr\u00e9\u00e9 avant l’ex\u00e9cution de cette t\u00e2che. La propri\u00e9t\u00e9 from<\/strong> prend un fichier du stockage interne (en r\u00e9alit\u00e9 l’URI du fichier, celui-ci sera r\u00e9cup\u00e9r\u00e9 au dernier moment depuis le stockage interne). J’utilise ici l’expression Pebble {{inputs.beers}}<\/code> pour r\u00e9cup\u00e9rer le fichier pass\u00e9 en input au flow.<\/p>\nio.kestra.plugin.gcp.bigquery.Query<\/code> permet d’effectuer une requ\u00eate BigQuery. Il y a plusieurs mani\u00e8res de stocker le r\u00e9sultat de la requ\u00eate. Ici, j’utilise la propri\u00e9t\u00e9 fetchOne: true<\/code> qui configure la t\u00e2che pour r\u00e9cup\u00e9rer une seule ligne et mettre le r\u00e9sultat de la requ\u00eate en output de la t\u00e2che. Il est aussi possible de charger toutes les lignes (fetch: true<\/code>), ou de stocker les r\u00e9sultats dans le stockage interne de Kestra (store: true<\/code>) ce qui est conseill\u00e9 pour les requ\u00eates qui ram\u00e8nent beaucoup de lignes.<\/p>\n{{outputs['query-beers'].row.count}}<\/code>.<\/p>\n{{outputs['query-beers'].row.count}}<\/code> peut sembler intrigante :<\/p>\noutputs['query-beers']<\/code> : signifie l’output de la t\u00e2che query-beers<\/strong>, nous avons jusqu’\u00e0 pr\u00e9sent vu la notation point\u00e9e (.) pour acc\u00e9der aux outputs, mais quand celui-ci contenant un caract\u00e8re ‘-‘, on est oblig\u00e9 d’utiliser la notation subscript ([]) car ‘-‘ est un caract\u00e8re sp\u00e9cial pour Pebble.<\/li>\n\nrow<\/code> : est le nom de l’attribut mis en output dans la t\u00e2che query-beers<\/strong>, une t\u00e2che peut avoir plusieurs attributs en outputs, se r\u00e9f\u00e9rer \u00e0 la documentation de la t\u00e2che pour leur liste.<\/li>\n\ncount<\/code> : est le nom de la colonne.<\/li>\n<\/ul>\nForEach et format de fichier<\/h2>\n
\nid: wikipedia-top-ten\nnamespace: fr.loicmathieu.example\ndescription: A flow that loads wikipedia top 10 FR pages each hour\n\ntasks:\n - id: start-log\n type: io.kestra.core.tasks.debugs.Echo\n format: \"{{taskrun.startDate}} - {{task.id}} - Launching the flow\"\n - id: for-each-countries\n type: io.kestra.core.tasks.flows.EachSequential\n tasks:\n - id: query-top-ten\n type: io.kestra.plugin.gcp.bigquery.Query\n serviceAccount: \"\"\n sql: |\n SELECT DATETIME(datehour) as date, title, views \n FROM `bigquery-public-data.wikipedia.pageviews_2023` \n WHERE DATE(datehour) = current_date() and wiki = '{{taskrun.value}}' and title not in ('Cookie_(informatique)', 'Wikip\u00e9dia:Accueil_principal', 'Sp\u00e9cial:Recherche')\n ORDER BY datehour desc, views desc\n LIMIT 10\n store: true\n - id: write-csv\n type: io.kestra.plugin.serdes.csv.CsvWriter\n from: \"{{outputs['query-top-ten'][taskrun.value].uri}}\"\n value: '[\"fr\", \"en\", \"de\"]'\n<\/pre>\nio.kestra.core.tasks.flows.EachSequential<\/code> permet de faire une boucle. Elle va r\u00e9aliser plusieurs fois la liste des t\u00e2ches enfant pour les valeurs pass\u00e9es dans sa propri\u00e9t\u00e9 value<\/strong> ; ici les langues fr, en et de.<\/p>\nio.kestra.plugin.gcp.bigquery.Query<\/code> va ex\u00e9cuter une requ\u00eate sur BigQuery qui sera stock\u00e9e dans le stockage interne de Kestra (store: true<\/code>). Elle utilise l’expression Pebble {{taskrun.value}}<\/code> qui permet de r\u00e9cup\u00e9rer la valeur en cours de la boucle EachSequential<\/code>.<\/p>\nio.kestra.plugin.serdes.csv.CsvWriter<\/code> va r\u00e9\u00e9crire le fichier stock\u00e9 par la t\u00e2che pr\u00e9c\u00e9dente au format CSV. Par d\u00e9faut, Kestra utilise le format de stockage objet Amazon Ion, cette t\u00e2che permet donc de passer du format Ion au format CSV. Elle utilise l’expression Pebble {{outputs['query-top-ten'][taskrun.value].uri}}<\/code> dont l’attribut [taskrun.value].uri<\/code> permet de r\u00e9cup\u00e9rer la valeur de l’attribut uri<\/strong> pour l’it\u00e9ration de boucle actuelle.<\/p>\n![\"\"](\"https:\/\/i0.wp.com\/loicmathieu.fr\/wordpress\/wp-content\/uploads\/kestra-04.png?resize=640%2C468&ssl=1\")
\nTrigger<\/h2>\n
\nid: flow-with-trigger\nnamespace: fr.loicmathieu.example\n\ntriggers:\n - id: schedule\n type: io.kestra.core.models.triggers.types.Schedule\n cron: \"*\/1 * * * *\"\n\ntasks:\n - id: \"echo\"\n type: \"io.kestra.core.tasks.debugs.Echo\"\n format: \"{{task.id}} > {{taskrun.startDate}}\"\n<\/pre>\nTraitement de donn\u00e9e avec Python et Pandas<\/h2>\n
io.kestra.core.tasks.scripts.Python<\/code> qui permet de lancer un script Python.<\/p>\ninputFiles<\/code> : une liste de fichier qui doit contenir le fichier main.py<\/code> qui sera appel\u00e9 par la t\u00e2che. Un deuxi\u00e8me ficher data.csv<\/code> est d\u00e9fini qui va permettre d’acc\u00e9der localement au fichier cr\u00e9\u00e9 par la t\u00e2che write-csv<\/strong>. Kestra va automatiquement le r\u00e9cup\u00e9rer depuis son internal storage et le mettre \u00e0 disposition dans le r\u00e9pertoire de travail de la t\u00e2che Python.<\/li>\n\nrequirements<\/code> : une liste de d\u00e9pendances pip, ici, nous avons mis la librairie Pandas qui permet d’analyser les donn\u00e9es du fichier CSV.<\/li>\n<\/ul>\n\nid: wikipedia-top-ten-pyhton-panda\nnamespace: fr.loicmathieu.example\ndescription: A flow that loads wikipedia top 10 FR pages each hour\n\ntasks:\n - id: query-top-ten\n type: io.kestra.plugin.gcp.bigquery.Query\n serviceAccount: \"\"\n sql: |\n SELECT DATETIME(datehour) as date, title, views \n FROM `bigquery-public-data.wikipedia.pageviews_2023` \n WHERE DATE(datehour) = current_date() and wiki = 'fr' and title not in ('Cookie_(informatique)', 'Wikip\u00e9dia:Accueil_principal', 'Sp\u00e9cial:Recherche')\n ORDER BY datehour desc, views desc\n LIMIT 10\n store: true\n - id: write-csv\n type: io.kestra.plugin.serdes.csv.CsvWriter\n from: \"{{outputs['query-top-ten'].uri}}\"\n - id: \"python\"\n type: io.kestra.core.tasks.scripts.Python\n inputFiles:\n data.csv: \"{{outputs['write-csv'].uri}}\"\n main.py: |\n import pandas as pd\n from kestra import Kestra\n data = pd.read_csv(\"data.csv\")\n data.info()\n sumOfViews = data['views'].sum()\n Kestra.outputs({'sumOfViews': int(sumOfViews)})\n requirements:\n - pandas\n<\/pre>\n![\"\"](\"https:\/\/i0.wp.com\/loicmathieu.fr\/wordpress\/wp-content\/uploads\/kestra-05.png?resize=640%2C259&ssl=1\")
\nConclusion<\/h2>\n