---
title: "Airflow FMC walkthrough"
canonical: "https://docs.vaultspeed.com/space/VPD/3038412824/Airflow%20FMC%20walkthrough"
format: markdown
---
> Macro (toc)

### Parameters

Start by configuring the FMC parameters.

  FMC_GENERATE_SRC_LOADING_SCRIPTS: If enabled, extra tasks will be added to the generated workflow that will transfer data from your source system to the data warehouse.  
This is only used when not using an ETL tool. Otherwise, this should be set up in your ETL tool.  
configurations:
  - initial load:
    - source table → INI table
  - Full Load:
    - source table → CDC table (all data will be transferred)
  - Full Load using modification date:
    - source table → CDC table (all data will be transferred)
  - Incremental Load using modification date:
    - source tale → CDC table (only the latest changed data will be loaded)
  - Incremental Load using modification date & Flag: (remote journalling, otherwise the CDC system is expected to handle the transfer of data):
    - transfer to the firewall table through a database link in the mapping (if SQL, then it will select from a table through a database link if ELT tool then mapping from src to target)
    - source CDC table → CDC (only data in the window will be loaded)
    - source delta view → target delta table (requires the creation of views on the CDC tables & the creation of a table to store the delta window on the source system)
- FMC_SKIP_COMPLETED_TASKS: Indicates whether tasks in the FMC are run only once if they were successful or if all tasks are rerun.   
This creates a more complicated workflow (and slightly slower runs because there are more tasks) but can save time by not rerunning tasks even when restarting an entire run instead of specific tasks. (see the paragraph on handling failures)  
This functionality, when added, can be disabled for some runs by changing the rerun_all_tasks parameter in airflow.
- FMC_RERUN_ALL_TASKS, FMC_BATCH_SIZE: these parameters have no effect on the generated code but will be put into the JSON file containing the airflow variables, which can then be imported into Airflow and changed there. (see the “variables” section of this page: [https://vaultspeed.atlassian.net/wiki/pages/createpage.action?spaceKey=vpd&title=Airflow%20Installation%20Instructions&linkCreation=true&fromPageId=3038412824](https://vaultspeed.atlassian.net/wiki/pages/createpage.action?spaceKey=vpd&title=Airflow%20Installation%20Instructions&linkCreation=true&fromPageId=3038412824) ).
- SCHEMA_PL_PROCEDURES: schema where your Presentation Layer procedures (or additional business vault procedures) are stored. This is used when generating code for loading the BV & PL using SQL instead of an ELT tool.
- SOURCE_PL_DEP: determines whether the load of the BV & PL will wait for the load of this source to be successful or completed (the BV load will never run if there is still a running load for one of the sources in your DV).
- USE_FMC: Whether you will use our FMC will cause some extra objects to be generated with your DDL and ETL.
- FMC_DYNAMIC_LOADING_WINDOW: Whether the beginning of the loading window is determined by the end of the last successful run or whether the load will have a static loading window but will wait for the successful execution of the previous run.  
See the “Workflow logic” section of this page: [https://vaultspeed.atlassian.net/wiki/pages/createpage.action?spaceKey=vpd&title=Airflow%20Installation%20Instructions&linkCreation=true&fromPageId=3038412824](https://vaultspeed.atlassian.net/wiki/pages/createpage.action?spaceKey=vpd&title=Airflow%20Installation%20Instructions&linkCreation=true&fromPageId=3038412824) for more info.
- FMC_OVERLAPPING_LOADING_WINDOWS: If this is enabled, the start date of your loading window will be moved backward by an amount specified by FMC_WINDOW_OVERLAP_SIZE and FMC_WINDOW_OVERLAP_TYPE. This will be stored into an Airflow variable so it can be modified later.
- FMC_TYPE: The FMC type to use, Airflow or Azure Data Factory([Azure Data Factory FMC](https://vaultspeed.atlassian.net/wiki/spaces/VPD/pages/3012788295)) or Matillion(WIP).
- OBJECT_SPECIFIC_LOADING_WINDOW: Determines wether the loading window will be for an entire source, or specific per object (more info here: [https://vaultspeed.atlassian.net/wiki/spaces/VPD/pages/3081633856](https://vaultspeed.atlassian.net/wiki/spaces/VPD/pages/3081633856))
- CDC_BASED_LOADING_WINDOW: use the cdc-timestamp available in the cdc-tables to determine the loading window (more info here: [https://vaultspeed.atlassian.net/wiki/spaces/VPD/pages/3081633856](https://vaultspeed.atlassian.net/wiki/spaces/VPD/pages/3081633856))
- FMC_FILE_EXTENSION: extension for the files when deployed to notebooks in Databricks.

### Code generation

> 📝 Make sure you first have generated (and preferable deployed) your DDL and ETL code.

Navigate to the Flow Management Control screen, using the main menu in VaultSpeed. 

To create a new workflow, click the ‘+ FMC workflow’ button.

![image](media://f16a71e2-da8b-4c06-a767-4fd06abca35b)

![image](media://7f0c390c-9ace-42f3-9e5c-cb1d9b3c47b1)

![image](media://cd5dffec-d754-4790-b9de-1c6835ada5d9)

First, we will create a workflow for the initial load

- Select FL (foundation layer, a.k.a Raw Data Vault)
- Select INIT as the load type.
- Select your Data Vault.
- Select one of your sources.
- Choose a DAG name. A Directed Acyclic Graph (DAG) is what Airflow calls a workflow. A workflow is a graph or network of tasks with dependencies between tasks (Directed) and without circles (Acyclic).  
This name should be unique to this workflow, e.g., <dv name>_<source name>_init.

> ⚠️ Your DAG name can NOT start with a number, this would generate invalid python DAG code.

- Add a description for your flow.
- Choose a start date, this defines the initial load date. This date can be in the past if you want to do historical loads using CDC data. This date also defines the start of your loading window for the incremental loads, so this could be the date of the database dump for the initial data.

> ⚠️ This date is set in the same timezone as your Airflow installation, by default UTC.  If you want to set a specific datetime, take into account the conversion to UTC first if needed.

- Set the concurrency, i.e., how many tasks in this workflow are allowed to be executed simultaneously. This depends on the power of your target database and/or ETL tool and the power of your airflow schedulers and workers.
- Enable group tasks when using SQL code and mini-batches. This reduces the overhead of creating connections to the DB. So if connecting takes a lot longer than the actual mapping execution, enabling this option could be useful.

> ⚠️ This option does not work when using ETL tools (Talend and ODI) or Spark SQL.

- Enter a name for your source database connection if you are using source loading. This connection will be defined in Airflow later.
- Enter a name for your data vault database connection.  This connection will be defined in Airflow later.
- When using Talend, ODI, or Spark SQL, or Datbricks, then the ETL connection has to be filled in as well(see the “connections” section of this page: [https://vaultspeed.atlassian.net/wiki/pages/createpage.action?spaceKey=vpd&title=Airflow%20Installation%20Instructions&linkCreation=true&fromPageId=3038412824](https://vaultspeed.atlassian.net/wiki/pages/createpage.action?spaceKey=vpd&title=Airflow%20Installation%20Instructions&linkCreation=true&fromPageId=3038412824)).

Next, we will create an incremental load workflow. Click the "+ FMC workflow" button again and select FL.

- Select INCR for load type.
- Select the same DV and source.
- Choose a DAG name e.g. <dv name>_<source name>_incr.

> ⚠️ Your DAG name can NOT start with a number, this would generate invalid python DAG code.

- Add a description.
- Choose the same start date as your incremental load. That way, the initial load will run on that date, and the incremental load will start at start_date+interval and load all data between start_date and start_date + interval.
- Set the concurrency, i.e., how many tasks in this workflow are allowed to be executed simultaneously. This depends on the power of your data warehouse database and the power of your airflow schedulers and workers (which can easily be changed later).
- Enable group tasks when using SQL code and mini-batches. This reduces the overhead of creating connections to the DB. So if connecting takes a lot longer than the actual mapping execution, enabling this option could be useful.

> ⚠️ This option does not work when using ETL tools (Talend and ODI) or Spark SQL.

- Set a schedule_interval. This is the time between incremental loads. The allowed values are:
  - "@hourly": Run once an hour at the beginning of the hour (this can not be used for the BV loads).
  - "@daily": Run once a day at midnight (this can not be used for the BV loads).
  - "@weekly": Run once a week at midnight on Sunday morning (this can not be used for the BV loads).
  - "@monthly": Run once a month at midnight of the first day of the month (this can not be used for the BV loads).
  - "@yearly": Run once a year at midnight of January 1 (this can not be used for the BV loads).
  - cron expression: [https://en.wikipedia.org/wiki/Cron#CRON_expression](https://en.wikipedia.org/wiki/Cron#CRON_expression)  (e.g. "0 0 * * *")
  - python timedelta() function : [https://docs.python.org/3/library/datetime.html#datetime.timedelta](https://docs.python.org/3/library/datetime.html#datetime.timedelta)  (e.g. timedelta(*minutes=15*))

> ⚠️ The double quotes need to be included when entering the values in the interval field!

- Enter a name for your source and data vault database connections. These connections will be defined in Airflow later.

These settings can be changed later by editing the created workflows in the FMC screen using the Actions menu.

![image](media://b9be72be-4ea7-45ac-95e2-ec592e0e8201)

To generate code for the workflows, select one of them and use the options in the Actions menu to either Generate code based on the latest ETL generation, of just use the Generate option to be able to choose based on which ETL generation you want to create the code.

![image](media://50184c4a-4243-4c16-ac25-ba9204df2939)


When you select ‘Generate’, you get a screen with all ETL generations related to the source that is attached to this flow.  Select the ETL generation where you want your flow to be based on and click the ‘Start Generation’ button.

![image](media://7265a5d5-48c9-446e-a666-26b6d35dd1e6)


This will launch new tasks, one to generate the code and one for the agent to download.

> ℹ️ If you clicked the ‘Generate the Latest' button, it will automatically start the tasks based on your latest ETL generation.

Selecting a workflow and clicking "View generations" will show you all your workflow generations, with their settings at that time, for which data vault and source release it was generated, when they were generated, and the name of the zip file containing the generated code.

![image](media://3f3d1b2d-1e0b-464b-834d-264600b29c7f)

### Airflow set up

Install and set up Airflow, see [Airflow installation instructions](https://vaultspeed.atlassian.net/wiki/pages/createpage.action?spaceKey=vpd&title=Airflow%20Installation%20Instructions&linkCreation=true&fromPageId=3038412824).

Start the airflow webserver and scheduler.

Open the Airflow web portal in your browser.

![image](media://c6e0e5c3-9c3b-4fed-ad77-0c66b8ba66a1)

If the plugin was installed correctly, you should see a VaultSpeed menu item at the top. For now, this only contains a link to VaultSpeed and the FMC documentation, but we will add more in the future (e.g., extra monitoring and load info charts).

Click on admin and connections and click the plus symbol to define a new connection.

![image](media://697304ac-aa6b-4643-be1b-12bff14f4d7d)

If the plugin provider package was successfully installed, then you should also see our custom connection types: ODI, Spark SQL, and Talend.

![image](media://680ad764-1f41-4b20-b9ef-bf8540ee72c2)

Fill in the fields for your source and DV connections, conn id should be the same as the connection name entered in the accelerator (these connections will be encrypted and stored in the Airflow metadata database).

![image](media://ff59cf84-e5b6-401f-ad58-25fb8d35445b)

Click on Admin and Variables.  
Add the following variables, or select choose file and select the variables file included in the generated FMC code and click import variables.

![image](media://9c203e03-a8af-4505-bb43-705b1ce1d5c8)

Unzip the generated code for the initial and incremental load and move the python files to the airflow_home/dags directory and the JSON files to the directory defined in the path_to_metadata_variable (this can be the same directory).  
In this case, all the files were placed inside the /home/airflow/dags directory.

Return to the Airflow web page and go to the DAGs menu, you should see the following (with different values of course):

![image](media://83be67e3-420a-47ce-87b4-d1a36109adac)

If you don't see this, wait a bit (the webserver only checks for new DAGs once in a while depending on your settings) or restart the webserver and scheduler. If there was a problem importing the DAGs, it will be shown in the UI and can also be found in the import_error table in the Airflow metadata database.  


### Starting the workflows

To start executing the loads, we need to unpause the workflow, click the switch at the left of the DAGs table.

![image](media://6253c4bb-2546-4a7a-b279-51133e4b44c0)

Your DAG will start running if the current time is past the start date in case of the initial loads. The incremental loads will start after the start date + interval.

### monitoring the workflows

![image](media://012e477a-59dc-493b-8483-1c99bf0c6606)

Here you can see an overview of the status of the tasks in each workflow and the workflows themselves, here we have 2 completed loads and 1 active load with 4 tasks that are running. Under actions, you can trigger the DAG to run immediately, regardless of its schedule. The refresh button will trigger the dag file to be parsed again and updated. The delete button will remove the DAG and all its metadata.  


Clicking on a DAG name will open the Graph View. This view only shows one run (can be chosen in the dropdown) and is ideal to track the progress of an active run.

![image](media://4ee33e71-1ccc-48df-8936-375b961455ef)

  
In the Tree View, you can see the status of all tasks in the last runs, for every run a column of squares will appear, this view is mainly useful to compare between runs, which runs are shown can be controlled by setting the date and the number of runs.

![image](media://cb76dfed-442e-49bb-9e04-19b5445f8e75)

 

In both these views, hovering over a task will reveal some details such as start time, end time, and duration.  
Clicking on a task will show the task controller:  


![image](media://529c507f-57b0-4a3b-89c1-8211af2986a8)

  
Clicking Log will show you the log for that task.

![image](media://a44fbf93-5267-4c94-a2d8-e57cf2f1d47f)

  
Task instance details show you all the details of that task.

![image](media://86e606de-0337-4c21-b59a-5b93e5b3d4e7)

  
Rendered will show you what the task executed (e.g. the sql statement or bash command). This is mainly useful for templated tasks such as the fmc_mtd tasks.

![image](media://c4bb9f27-1bb1-404e-b5a3-10fe12432357)

  
All Instances will take you to the Task instances menu where you can see a table with all historic executions of that same task.

In the Task Actions part, you can also (re)run the task manually, clear the task (and dependencies, controlled by the buttons next to it) such that it gets rescheduled, or set the task (and dependencies, controlled by the buttons next to it) to success or failed.  
The Run button can only be used with the Celery Executor.

In the Calendar view, you can find an overview of the runs per day in a calendar.

![image](media://467ed9fa-6934-4e9d-ad9a-ab41c521b33e)

 

Clicking on Task durations will show a chart of task duration over the past runs (clicking on the circle next to a task will remove it from the chart and double-clicking will remove everything except that task.

![image](media://89965689-4430-4a82-b717-6f40ecbb9f40)

 

Task tries is a similar-looking graph, except that it shows how many tries it took before the task was successful.

Landing times is again a similar chart, this one shows the time difference between when the task was supposed to run (its schedule) and when it was completed.  


Gantt will show a Gantt chart, this chart shows when the tasks for a certain run started and ended.

![image](media://975dd48f-b3e3-46f5-97cc-9fc9025534f2)

Details will show the details of a DAG such as concurrency, file location, tasks, etc.

Code will show u the python file that defined this DAG.

All of these screens are also accessible through the links at the right side of the DAGs in the DAGs view.

 

### Reporting and monitoring

All the flow metadata for the VaultSpeed code is stored in the schema_fmc.fmc_loading_history table. It contains the flow name, the source business key(DV name for BV flows), the load date and id, the loading window, and the start and end-time of the load and its status.  
Success flag = 1 means completed successfully, 0 means failed and null means that it is still running.

![image](media://24dd8c7c-d199-48ad-9a83-97ff7d05b34c)

The Load Cycle Id is the same as the dag_id in the Airflow metadata database’s dag_run table. This can be used to join the Airflow data and the DV data and metadata. The load date is the same as the execution_date.

![image](media://8fe87b30-d5dc-4b4d-b727-2fa5698ffcf7)