We will see these concepts in practice in the following article.<\/p>\n
To launch Kestra locally, the easiest way is to use the Docker compose provided<\/a> file, then run it via This Docker compose will run Kestra in standalone<\/strong> mode (all components in a single process) with a PostgreSQL database as a queue and repository (for storing and executing flows), and a local storage (for storing flow data).<\/p>\n Kestra is composed of several components that communicate together via queues and store flow information via repositories, these queues and repositories can be implemented in different ways (memory, JDBC, Kafka, Elasticsearch). I’m not going into detail about Kestra’s architecture today, but for the more curious you can refer to the architecture documentation<\/a> of Kestra.<\/p>\n Once Kestra is started, a graphical interface will be available on port 8080: http:\/\/localhost:8080<\/a><\/a>, it’s via this interface that we’ll do all the examples in this article.<\/p>\n At the first launch, a Guided Tour will be offered, you can follow it or skip it to be able to follow the examples in this article.<\/p>\n To create a flow, go to the Flows<\/strong> menu and then click on the Create<\/strong> button at the bottom right. Now you have a textarea in which you will be able to enter the YAML description of the flow.<\/p>\n A flow has:<\/p>\n Here we have added a single task that has three properties:<\/p>\n Each task will have its own properties, which are documented in the online documentation as well as in the documentation integrated in the Kestra graphical interface (Documentation -> Plugins -> then select a plugin in the right menu).<\/p>\n Within the editor, the YAML description of the flow is validated and autocompletion is available for the type and properties of the tasks via CTRL+SPACE.<\/p>\n Kestra is based on a plugin system that allows you to add tasks (but also triggers or conditions that we will see later). By default, there are very few plugins provided in Kestra but the Docker image used in the Docker compose is built by adding all the plugins maintained by the Kestra development team, their documentation is here : https:\/\/kestra.io\/plugins\/<\/a><\/a>.<\/p>\n To launch this first flow, go to the Executions<\/strong> tab and then click the New Execution<\/strong> button, which will switch to the Gantt view of the flow execution which is updated in real time with the status of the execution.<\/p>\n The Logs<\/strong> tab allows you to see the runtime logs.<\/p>\n Note the Now let’s add an input named name<\/strong> to our flow to give it the name of the person to say hello to. To do this, we need to edit the flow by selecting the flow, either via the breadcrumb navigation at the top ( Home \/ Flows \/ fr.loicmathieu.example.hello-world), or by left-clicking on Flows<\/strong> and selecting it from the list; then go to the Source<\/strong> tab.<\/p>\n We can pass data to a flow via inputs. We will define a name<\/strong> input of type STRING<\/strong>.<\/p>\n When we click on the New Execution button, a form asks us to fill in the flow inputs.<\/p>\n After entering a value for the name and running the flow, we can see that the log contains the value of the input.<\/p>\n In this example, the flow will take a CSV file as input, load it into Google BigQuery, then run a query on the loaded data and display the result in the logs.<\/p>\n I’m using BigQuery here because it’s easier than starting a database, but it requires a Google Cloud account and a service account set up. To do this, you need to create a Google Cloud service account<\/a> and use it in the Without further ado, the solution ;)<\/p>\n Here I have defined an input of type The load-beers<\/strong> task of type The query-beers<\/strong> task of type The end-log<\/strong> task will write a log, we have already seen it before. Here, we want to write in the log the number of records loaded from the database, so we will get the corresponding output from the query-beers<\/strong> task via a Pebble expression: The expression In this example, we will query the 10 most viewed Wikipedia pages using the BigQuery public dataset wikipedia.pageviews_2023<\/strong> for French, English and German languages. Then we will transform the result into CSV.<\/p>\n The for-each-countries<\/strong> task of type The query-top-ten<\/strong> task of type The write-csv<\/strong> task of type After executing the flow, you can go to the Outputs<\/strong> tab to access the task outputs and, among other things, download the CSV files generated by the flow.<\/p>\n By default, a flow can only be runned manually via the graphical interface or the Kestra API.<\/p>\n It is possible to trigger a flow from an external event, this is the role of the trigger.<\/p>\n Kestra includes three basic triggers: flow<\/strong>, which allows you to trigger a flow from another flow, webhook<\/strong> which allows you to trigger a flow from a webhook URL and schedule<\/strong> which allows you to trigger a flow periodically from a cron expression. Many other triggers are available within Kestra’s plugins and allow to trigger a flow from a message in a broker, a file, or the presence of a record in a database table for example.<\/p>\n The following example will trigger a flow every minute, it uses a cron expression to define its triggering periodicity.<\/p>\n Kestra offers advanced functionality via tasks that allow you to execute Bash, Python or NodeJS scripts.<\/p>\n The following example will query the same BigQuery dataset of Wikipedia page views, write it to CSV format, and then use that CSV in a This task takes as properties:<\/p>\n The Python script will use Pandas to read the CSV file and transform it into a Pandas data frame, then perform the sum of the views<\/strong> column. This sum will then be put into the task output using the Python Kestra library.<\/p>\n Here are the execution logs of this flow.<\/p>\n In this introductory article, we have seen the main concepts of Kestra and some flow examples.<\/p>\n To go further, you can check out the online documentation<\/a> as well as the plugin list<\/a>.<\/p>\n Ketra is an open source community project available on GitHub, feel free to:<\/p>\n <\/p>","protected":false},"excerpt":{"rendered":" Kestra is an open-source data orchestrator and scheduler. With Kestra, data workflows, called flows, use the YAML format and are executed by its engine via an API call, the user interface, or a trigger (webhook, schedule, SQL query, Pub\/Sub message, …). The important notions of Kestra are : The flow: which describes how the data will be orchestrated (the workflow thus). It is a sequence of tasks . The task: a step in the flow that will perform an action…docker compose up<\/code>.<\/p>\nMy first 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> property that is its unique identifier within a namespace.<\/li>\n\nnamespace<\/code> property, namespaces are hierarchical like directories in a file system.<\/li>\n\ntasks<\/code> property which is the list of tasks to be performed at the execution of the flow.<\/li>\n<\/ul>\nid<\/code> property that is its unique identifier within the flow.<\/li>\n\ntype<\/code> property which is the name of the class that represents the task.<\/li>\n\nformat<\/code> property, this is a property specific to Echo tasks that defines the format of the message to be logged (an Echo task is like the echo shell command).<\/li>\n<\/ul>\n![\"\"](\"https:\/\/i0.wp.com\/loicmathieu.fr\/wordpress\/wp-content\/uploads\/kestra-01-1024x267.png?resize=640%2C167&ssl=1\")
\n![\"\"](\"https:\/\/i0.wp.com\/loicmathieu.fr\/wordpress\/wp-content\/uploads\/kestra-02-1024x270.png?resize=640%2C169&ssl=1\")
\nHello World<\/code> log generated by the hello<\/strong> task.<\/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-1024x262.png?resize=640%2C164&ssl=1\")
\nHello {{inputs.name}}<\/code> is an expression that uses the Pebble templating engine that will render what is between the {{<\/code> and }}<\/code> moustache. Here the Pebble expression will fetch the value of the inputs.name<\/strong> variable that points to the input with name name<\/strong>.<\/p>\nInputs and BigQuery<\/h2>\n
serviceAccount<\/code> property of the BigQuery task. To avoid hard-coding this variable, it is possible to use an environment variable or a cluster global variable, see the online documentation on 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> named beers<\/strong>. Kestra will detect the type of the input and generate the corresponding flow execution form. This will upload a CSV file containing a list of beers. This file will be stored in Kestra’s internal storage and available in all tasks.<\/p>\nio.kestra.plugin.gcp.bigquery.Load<\/code> is used to load a CSV file into a BigQuery table. The BigQuery dataset beers<\/strong> must have been created before this task is executed. The from<\/strong> property takes a file from the internal storage (actually the URI of the file, it will be retrieved at the last moment from the internal storage). Here I use the Pebble expression {{inputs.beers}}<\/code> to retrieve the file passed as input to the flow.<\/p>\nio.kestra.plugin.gcp.bigquery.Query<\/code> is used to perform a BigQuery query. There are several ways to store the query result. Here, I use the fetchOne: true<\/code> property which configures the task to fetch a single row and put the result of the query in the task output. It is also possible to load all rows (fetch: true<\/code>), or store the results in Kestra’s internal storage (store: true<\/code>) which is recommended for queries that bring back many rows.<\/p>\n{{outputs['query-beers'].row.count}}<\/code>.<\/p>\n{{outputs['query-beers'].row.count}}<\/code> may seem intriguing:<\/p>\noutputs['query-beers']<\/code>: means the output of the query-beers<\/strong> task, so far we have seen the dotted notation (.) to access outputs, but when this one containing a ‘-‘ character, we are forced to use the subscript notation ([]) because ‘-‘ is a special character for Pebble.<\/li>\n\nrow<\/code>: is the name of the attribute set as output in the query-beers<\/strong> task, a task can have multiple attributes as outputs, refer to the task documentation for their list.<\/li>\n\ncount<\/code>: is the name of the column.<\/li>\n<\/ul>\nForEach and file format<\/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> allows for looing. It will perform the list of child tasks several times for the values passed in its value<\/strong> property; here the languages fr, en and de.<\/p>\nio.kestra.plugin.gcp.bigquery.Query<\/code> will execute a query on BigQuery that will be stored in Kestra’s internal storage (store: true<\/code>). It uses the Pebble expression {{taskrun.value}}<\/code> which retrieves the current value from the EachSequential<\/code> loop.<\/p>\nio.kestra.plugin.serdes.csv.CsvWriter<\/code> will rewrite the file stored by the previous task to the CSV format. By default, Kestra uses the Amazon Ion object storage format, so this task switches from the Ion to the CSV format. It uses the Pebble expression {{outputs['query-top-ten'][taskrun.value].uri}}<\/code> whose [taskrun.value].uri<\/code> attribute retrieves the value of the uri<\/strong> attribute for the current loop iteration.<\/p>\n![\"\"](\"https:\/\/i0.wp.com\/loicmathieu.fr\/wordpress\/wp-content\/uploads\/kestra-04-1024x749.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>\nData processing with Python and Pandas<\/h2>\n
io.kestra.core.tasks.scripts.Python<\/code> task that allows you to run a Python script.<\/p>\ninputFiles<\/code>: a file list that should contain the main.py<\/code> file that will be called by the task. A second file data.csv<\/code> is defined that will allow local access of the file created by the task write-csv<\/strong>. Kestra will automatically retrieve it from its internal storage and make it available in the working directory of the Python task.<\/li>\n\nrequirements<\/code>: a list of pip dependencies, here we put the Pandas library that allows to analyze the data of the CSV file.<\/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-1024x414.png?resize=640%2C259&ssl=1\")
\nConclusion<\/h2>\n