--- title: "Airflow Installation Instructions - 2" canonical: "https://docs.vaultspeed.com/space/VPD/3038150663/Airflow%20Installation%20Instructions%20-%202" format: markdown --- > Macro (toc) ## Minimum requirements - 8GB RAM - 4 VCPUs - 100GB Disk Add more CPU's and RAM to run more concurrent tasks, or use Celery with multiple workers to scale horizontally, or use Kubernetes ([https://kubernetes.io/blog/2018/06/28/airflow-on-kubernetes-part-1-a-different-kind-of-operator/](https://kubernetes.io/blog/2018/06/28/airflow-on-kubernetes-part-1-a-different-kind-of-operator/) ). Another option is to use a SaaS solution such as [https://cloud.google.com/composer](https://cloud.google.com/composer) , [https://www.astronomer.io/](https://www.astronomer.io/) or [https://aws.amazon.com/managed-workflows-for-apache-airflow/](https://aws.amazon.com/managed-workflows-for-apache-airflow/). ## Installation Instructions ### Supported Versions To see which versions of VaultSpeed support for Airflow, please have a look at this page: [https://vaultspeed.atlassian.net/wiki/spaces/VPD/pages/3084025875](https://vaultspeed.atlassian.net/wiki/spaces/VPD/pages/3084025875) ### Run Airflow on Astronomer Cloud Installation and setup instructions for Astronomer can be found at [Astronomer](https://vaultspeed.atlassian.net/wiki/spaces/VPD/pages/3038937120). ### Install Airflow Using Standard Docker You can also use docker. For testing and development, you can use the standard dockers from Airflow: [https://airflow.apache.org/docs/apache-airflow/stable/start/docker.html](https://airflow.apache.org/docs/apache-airflow/stable/start/docker.html) For a production-ready Docker image, go to [https://airflow.apache.org/docs/apache-airflow/stable/production-deployment.html#production-container-images](https://airflow.apache.org/docs/apache-airflow/stable/production-deployment.html#production-container-images) To run Airflow in a Kubernetes cluster, you can use the Helm chart from [https://airflow.apache.org/docs/helm-chart/1.0.0/](https://airflow.apache.org/docs/helm-chart/1.0.0/). Note that depending on your python installation you might have to use `pip3` instead of just `pip.` ### Install Airflow Locally To install airflow locally, follow the installation instructions at [https://airflow.apache.org/docs/apache-airflow/stable/installation.html](https://airflow.apache.org/docs/apache-airflow/stable/installation.html) These instructions are for Linux or MAC only. To run Airflow on a windows machine use Docker or WSL [https://docs.microsoft.com/en-us/windows/wsl/about](https://docs.microsoft.com/en-us/windows/wsl/about). #### Install the VaultSpeed Provider Depending on the sources or targets that you will use in the project, you have to install different dependencies for the VaultSpeed Provider. The provider is build such that you only need to install the dependencies that you actually use, so if your tatget platform is Snowglake, then you do not have to install the Google dependecies that would be required for BigQuery as the target platform. - Download the VaultSpeed Airflow plugin from the Flow Management Control page of VaultSpeed, this will give you a zip file containing the package. - Move the zip file to your Airflow server (or add it to the folder containing your docker compose file for Airflow). - Install the plugin: - adding `[all]` at the end will install all the dependencies for all platforms. - Adding `[databricks]` will install only the dependencies required for Databricks. - Adding `[google]` will install only the dependencies required for Google BigQuery. - Adding `[snowflake]` will install only the dependencies required for Snowflake. - Adding `[jdbc]` will install only the dependencies required for enabeling Source loading. - Any other target platform does not require additional dependencies in the provider, and in that case you do not have to add anything between brackets at the end. - When your Target platform is SQL Server, then install: `apache-airflow-providers-microsoft-mssql` If you are using docker, you can use the following extended dockerfile, which install the plugin during the build process: ```docker FROM apache/airflow:3.1.5 # Copy the plugin zip file to the image COPY vs_fmc_plugin.zip /tmp/vs_fmc_plugin.zip # Install the plugin globally RUN pip install --no-cache-dir "/tmp/vs_fmc_plugin.zip[all]" ``` In the Docker Compose file from airflow you then have to change the following part: comment the image line(6), and uncomment the build line(7): ``` x-airflow-common: &airflow-common # In order to add custom dependencies or upgrade provider distributions you can use your extended image. # Comment the image line, place your Dockerfile in the directory where you placed the docker-compose.yaml # and uncomment the "build" line below, Then run `docker-compose build` to build the images. #image: ${AIRFLOW_IMAGE_NAME:-apache/airflow:3.1.5} build: . env_file: - ${ENV_FILE_PATH:-.env} ``` #### Install a Database for Airflow Metadata Airflow requires a database to store its metadata. If you install airflow directly, you will have to set this up yourself. Airflow supports the following database types: - PostgreSQL (recommended) [https://airflow.apache.org/docs/apache-airflow/stable/howto/set-up-database.html#setting-up-a-postgresql-database](https://airflow.apache.org/docs/apache-airflow/stable/howto/set-up-database.html#setting-up-a-postgresql-database) - MySQL [https://airflow.apache.org/docs/apache-airflow/stable/howto/set-up-database.html#setting-up-a-mysql-database](https://airflow.apache.org/docs/apache-airflow/stable/howto/set-up-database.html#setting-up-a-mysql-database) ## Configure Airflow - change the airflow configuration settings by editing airflow.cfg in AIRFLOW_HOME, make sure to change the following settings: - executor type: Use LocalExecutor to run all tasks on the same machine as the scheduler, or CeleryExecutor to distribute the execution of task instances to multiple worker nodes [https://airflow.apache.org/docs/apache-airflow/stable/executor/celery.html](https://airflow.apache.org/docs/apache-airflow/stable/executor/celery.html). Note that the Sequential executor will not work properly for some configurations. - sql_alchemy_conn (connection to the metadata database, see [https://airflow.apache.org/docs/apache-airflow/stable/howto/set-up-database.html](https://airflow.apache.org/docs/apache-airflow/stable/howto/set-up-database.html), and for the connection string formats see [https://docs.sqlalchemy.org/en/13/core/engines.html](https://docs.sqlalchemy.org/en/13/core/engines.html) ) - load_examples = False - pool sizes, concurrency, and resource settings when needed - webserver host and port to allow access to the webserver from outside - celery settings when using it as an executor. - set allow_driver_path_in_extra=True and allow_driver_class_in_extra=True - set hide_sensitive_var_conn_fields = False - Reinitialize the airflow DB to the new connection as defined by the sql_alchemy_conn: - Create admin user(s), more users and roles can be added in the UI: There are other ways of authenticating users as well, see [https://airflow.apache.org/docs/apache-airflow/stable/security/webserver.html](https://airflow.apache.org/docs/apache-airflow/stable/security/webserver.html) - start your web server, and connect to it using your browser (http://host_name:8080): ## Running Airflow Airflow has multiple applications which need to be running, note that these do not have to be running on the same machine, and you can also run multiple instances of them for speed and redundancy: - webserver - scheduler, [https://airflow.apache.org/docs/apache-airflow/stable/scheduler.html#running-more-than-one-scheduler](https://airflow.apache.org/docs/apache-airflow/stable/scheduler.html#running-more-than-one-scheduler) - Worker (when using Celery) - Flower (when using Celery) These can be installed as services, or they can be managed by applications like [http://supervisord.org/](http://supervisord.org/).  ## Configure Airflow Connections In Airflow, go to Admin-Connections and Define the connections to your source (when using source loading) and data warehouse. > ℹ️ ### Security > ℹ️ > ℹ️ All connections and variables are stored encrypted in the Airflow metadata database.
SQL connection (Snowflake, BigQuery, Postgres, Oracle, SQL Server, Synapse, Singlestore) Set connection type to your tatget platform, and fill in all the required fields. The connection id should be set to the value entered in the DV connection name field of the workflow in VaultSpeed. You can find the detials about which fields are required, and the different authentication methods by looking at the Airflow documentations, for example: - [https://airflow.apache.org/docs/apache-airflow-providers-snowflake/stable/connections/snowflake.html](https://airflow.apache.org/docs/apache-airflow-providers-snowflake/stable/connections/snowflake.html) - [https://airflow.apache.org/docs/apache-airflow-providers-google/stable/connections/bigquery.html](https://airflow.apache.org/docs/apache-airflow-providers-google/stable/connections/bigquery.html) For BigQuery, make sure to fill in the location, the project_id and set use_legacy_sql to false (this is null by default, you can set it to false by editing the JSON, or toggeling it on and off with the button in the extra fields.)
Databricks connection Create a new Conection: - Set the connection type to Databricks. - fill in the host field with the URL to your Databricks instance - in the password field, enter your token - In the Extra Field JSON add the following properties: Add Databricks specific variables: - Create a variable called `databricks_path`, which contains the **absolute** path (starting and ending with `/`) to the root of the VaultSpeed notebooks (The same path as entered in the agent connection). - Create a Variable called `databricks_file_extension` which contains the extention of the deployed notebooks, this is normally either `.sql` when they are deployed via the VaultSpeed Agent or empty when they are created via a Git repository. You can double check the extention by looking into your Workspace in Databricks.
dbt connection There are 2 types of dbt connections available with the VaultSpeed provider: - dbt Cloud - dbt CLI The code generated by VaultSpeed works with both conection types, so you can, for example, use the CLI for local development and Cloud for production without needing to regenerate code.
source connection Set connection type to JDBC, and fill in all the fields. The connection id should be set to the value entered in the source connection name field of the workflow in VaultSpeed. In the driver path make sure to enter the full path to the JDBC driver for your database type (has to be downloaded separately). In the class name field, you need to fill in the class name for the specific JDBC driver (can be found by Googling: “ JDBC driver class name”). When defining a source JDBC connection for data loading you need to also add the path to the target JDBC driver. e.g. when loading from Postgres to Snowflake (no space before or after the comma!): jdbc_driver_loc = /home/airflow/jars/postgresql-9.4.1212.jar,/home/airflow/jars/snowflake-jdbc-3.8.8.jar This is needed due to an issue in the Python JDBC library.
JDBC connection (target for Source loading) Set connection type to JDBC, and fill in all the fields. The connection id should be set to the value entered in the DV connection name field of the workflow in VaultSpeed with “_jdbc” added at the end. In the driver path make sure to enter the full path to the JDBC driver for your database type (has to be downloaded separately). In the class name field, you need to fill in the class name for the specific JDBC driver (can be found by Googling: “ JDBC driver class name”). > ℹ️ When you are using source loading to a Bigquery environment, make sure you use bigquery JDBC driver version 1.2.21.1025. This to avoid an error on ‘TRUNCATE TABLE statement is not supported in transactions’.
Spark SQL connection There are 3 possible methods to run Spark SQL from Airflow using the VaultSpeed Provider, which method is used is determined by the type of the connection, the operator (and DAG code) stays the same. The generated code uses 2 connections, the dv_connection_name is used for the metadata queries, and the etl_connection_name is used to run the actual mappings. - Spark SQL CLI: - Set connection type to spark_sql_vs(Spark SQL VaultSpeed), and fill in all the fields. The connection id should be set to the value entered in the etl_connection_name field of the workflow in VaultSpeed. Spark task execution requires access to the Spark SQL CLI ([https://spark.apache.org/docs/latest/sql-distributed-sql-engine.html](https://spark.apache.org/docs/latest/sql-distributed-sql-engine.html) ). The "spark-sql" binary has to be in the PATH. - Spark JDBC: - Create 2 Spark JDBC connections (not Hive!) one with dv_connection_name as the conn_id, and one with the etl_connection_name. - Apache Livy: - Create 2 connections of the type spark_sql_livy(Spark Livy VaultSpeed), one with dv_connection_name as the conn_id, and one with the etl_connection_name. When using source loading, you must create a separate JDBC target connection, the connection should have “Conn Id” equal to the dv_connection name with “_jdbc” added at the end. > ℹ️ For connections that contain login and password (token) inside of the JDBC URL, these still need to be defined in connection’s login and password fields
### Airflow Variables The airflow variables need to exist when you want to run a DAG. The creation of these variables can be done by importing the JSON file that is generated by Vaultspeed (_variables_.json). See section “Generate ETL and DDL for how to generate that. The following variables will be created automatically during that import: - **path_to_metadata**: Path to the directory where the metadata JSON files are stored. - **path_to_sql**: Path to the directory where the generated SQL files are placed. This is only needed for target databases (Apache Spark SQL) that do not support stored procedures. In that case, the application will generate SQL files with the loading logic. - **window_overlap_<**`src_short_name`**>**: String containing the interval calculation SQL function with `` as the template variable for the window timestamp, e.g. ` + interval'15 minutes'`. This is only used if the corresponding VaultSpeed parameter is enabled. The values set in the window parameters will be used to create this variable in the generated variables JSON. - **batch_size**: When source loading scripts are generated, this is the number of records that will be loaded into the Target database simultaneously (set this dependent on the amount of memory and I/O available). ## Generate ETL and DDL Code. Enable the USE_FMC system parameter in VaultSpeed. This will add extra tables and metadata to the DDL and ETL generation. Generate code from the VaultSpeed Flow Management control page. Refer to the [Flow Management Control - FMC Flow: Generate](https://docs.vaultspeed.com/space/VPD/3696266471/Flow+Management+and+Control) section of the documentation for detailed steps. Place the generated python files in the `AIRFLOW_HOME/dags` directory (note that you can change the location of the dags directory in the airflow config) or a subdirectory, you might have to create this directory. Place the JSON files in a metadata directory. This metadata directory can be the dags directory or it can be a subdirectory. If you want to use our deploy to GIT feature, then this should be the dags directory or a subdirectory. Because in that case, both dag files and JSON files are placed in the same folder. e.g. `AIRFLOW_HOME/dags/vaultspeed` Execute the following command to check whether all the DAGs can be loaded: ``` (airflow) $airflow list_dags ``` Run the airflow scheduler (and workers when using Celery): ``` (airflow) $airflow scheduler ``` In Airflow go to DAGs and unpause the generated DAGs to allow them to run (switches in the first column, with :info:).  ## Usage Instructions ### Managing Failed Runs - If a run has failed on multiple tasks and u want to restart it after fixing the problem, go to the tree view for the DAG and click on the top red circle of the run u want to restart, click "clear", and "ok". This will clear the status of all the tasks for that run, and rerun all tasks in the dag. - If only a few tasks have failed, you can also rerun them by clicking on the failed tasks in the tree or graph view and clicking "clear" with downstream enabled (this is enabled by default). This will rerun only that task and its descendants. - If a task has failed and it is fixed without running the task, click on the failed task and next to "mark success," click "downstream", and then click "mark success". This will set the failed task and all its dependent tasks to success. Finally, clear the fmc_load_success task, which will update the FMC loading history table. ## Workflow Logic The FL workflow will be scheduled at a certain interval. If the FMC_DYNAMIC_LOADING_WINDOW parameter is enabled, then the load will be executed at the scheduled date with a loading window determined by the last successful load. This also means that your first incremental run will have a loading window starting with the start date of the initial load. So when you want to do a historic initial load (e.g. when you have CDC data from the last few months), set the start date of the initial workflow to a date in the past (e.g. 1 year ago). You can leave the init source tables empty to load only ghost records or fill them with data from a database backup taken at the start date. The first incremental load will then load all the historical data, starting from the init load start date until the start date of the incremental load + the interval. If the dynamic loading window logic is disabled, then each load will wait until the previous load is executed successfully. This means that the data won't be overwritten by the new data from your source (unless the failure was in the loading of the source data), and when u fix the issue and rerun the failed tasks, the data of the previous load will be added to the DV. Only then will the next load start and new data be loaded from the source. If more than one window passes before fixing the error, then you will still lose data if u do not have CDC. > ℹ️ The Business vault workflow will wait for all FL loads to be finished for sources where the `SOURCE_BV_DEP` parameter is enabled. After it detects that the FL loads are finished, it will check whether the last load for all sources was successful. If this is not the case, it will fail the BV load (to prevent incomplete PIT and bridge loading). Only if all dependent source loads were completed successfully, does the BV load start.   ## Loading the Presentation Layer (PL) VaultSpeed can also generate airflow code to run your mappings for loading extra BV objects and the PL. This is included in the logic for the BV workflows. To define which extra mappings should be executed and in which order, create the following JSON file in the defined metadata directory: "pl_mappings_"+pl_dag_name. This JSON file should contain a list of layers that will be executed one after the other. Each layer object should contain an array of mapping names. These will be executed in parallel. It is also possible to give an array of mapping names in this array instead of just mapping names. These will then be executed sequentially (see example below). example: ``` { "pl_stg":[["stg_a_1", "stg_a_3", "stg_a_2"],"srg_b","stg_c"], "pl":["pl1_a","pl_b","pl_c"], "ml":["pred_a"], "reports":["rep_a","rep_b"] } ``` result: ![image](media://ea8da1b8-9ac0-4d31-a211-fed696ff5341)