In the first part to this short series on Apache Airflow I outlined it its core architecture, installation process and created a sample example of a very rudimentary DAG (Directed Acyclic Graph). In this post I’d like to dive a bit deeper and look at a more advanced workflow which combines a few different operations and tasks. Let\u2019s assume that we\u2019re tasked with creating a DAG which needs to accomplish the following:<\/p>\n
- \n
- Programmatically create Citus database instance in Azure public cloud with specific configuration parameters e.g. geographic location, storage size, number of nodes assigned, firewall rules etc. These parameters are stored across two JSON template files, known Azure Resource Manager templates<\/li>\n
- Subsequently to creating Azure Citus database, the workflow should create a database schema using DDL SQL stored in one of the configuration files<\/li>\n
- Next, load text files data into the newly created database schema<\/li>\n
- Run four sample TPC-DS SQL queries in parallel (queries 5, 13, 47 and 57) and store the results into a file on the local file system<\/li>\n
- Shut the cluster down and remove any resources created as part of this process<\/li>\n
- Send out an e-mail notifying operator(s) of the workflow conclusion<\/li>\n<\/ul>\n
A similar type of workload is often used to create an idempotent pipeline which stitches together a number of tasks, including cloud infrastructure provisioning, data analytics, ETL\/ELT activities etc. to design an end-to-end data management solution. The required DAG, with all its dependencies and activities should like as per the image below.<\/p>\n
![\"\"](\"http:\/\/bicortex.com\/bicortex\/wp-content\/post_content\/\/2020\/04\/Airflow_Azure_Cloud_Job_DAG_Graph_View.png\")
<\/a><\/p>\nSample Airflow Azure Cloud Pipeline Architecture<\/h3>\n
In order to provision the Azure resource groups and resources, I have created two JSON template files, also known as Azure Resource Manager (ARM) templates, which contain the required resources and their properties. ARM templates allow for declarative creation and deployment of any Azure infrastructure components e.g. virtual machines, hosted databases, storage accounts etc. along their properties and names. As the libraries used in the Python version of the Azure SDK directly mirror and consume Azure service’s REST endpoints, it\u2019s quite easy to script out a repeatable deployment process using ARM templates as described in more details further along in this post. As part of the solution configuration, I have also created two SQL files which are used to create a suite of tables on the Citus cluster and run the queries, outputting the result sets into a local directory.<\/p>\n
![\"\"](\"http:\/\/bicortex.com\/bicortex\/wp-content\/post_content\/\/2020\/04\/Airflow_Solution_Files_Directory_View.png\")
<\/a>The aforementioned files used to populate the Citus cluster schema are part of the TPC-DS benchmark data (scaling factor set to 1GB) and consist of a number of flat files in a CSV format. As we’re not going to recreate the whole TPC-DS benchmark suite of tests (this post is not intended to evaluate Citus database performance), to simplify the data loading process, I will only use a few of them (just over 800MB in size), which should provide us with enough data to run a few queries once the tables have been populated. To simplify this demonstration, all Python solution files as well as ARM JSON templates, SQL files and flat files data were moved into dags directory aka {AIRFLOW_HOME}\/dags folder. Under typical scenario, data would not co-exists with configuration and solution files but for this exercise, I have nominated to structure the project as per the the directory breakdown on the left. Also, in a production environment, particularly one with many dag and task files, you would want to ensure that a consistent organizational structure is imposed across DAG folder and its sub-directories. As Airflow is such a versatile tool, depending on a specific use case, you may want to separate individual projects and their related hooks, operators, scripts and other files into their own respective folders etc.<\/p>\nWith Airflow server up and running (installation and configuration instructions in the previous post HERE<\/a>) and the files required to load into our Citus database schema staged in the solution directory we can start unpacking individual code samples used in the final DAG. As mentioned at the start, Citus database cluster, just as any other Azure resource, can be spun up declaratively using ARM template(s). You can peruse the content of the templates used in this demo as well as other artifacts in my publicly accessible OneDrive folder HERE<\/a>.<\/p>\n
A quick word about Citus (now owned by Microsoft). Citus is an open source extension to PostgreSQL which distributes data and queries across multiple nodes. Because Citus is an extension to PostgreSQL (not a fork), it gives developers and enterprises a scale-out database while keeping the power and familiarity of a relational database. Citus horizontally scales PostgreSQL across multiple machines using sharding and replication. Its query engine parallelizes incoming SQL queries across these servers to enable super-fast responses on even very large datasets. I’m not very familiar with the service but there are many posts on the Internet describing some interesting projects Citus was used in, beating other competing technologies (TiDB, Apache Pinot, Apache Kylin, Apache Druid), partly due to its great SQL support and a large community of PostgreSQL users who are already familiar with the RDBMS it is based on e.g. HERE<\/a>.<\/p>\n
![\"\"](\"http:\/\/bicortex.com\/bicortex\/wp-content\/post_content\/\/2020\/04\/Citus_Cluster_Simplified_Architecture.jpg\")
<\/a><\/p>\nNow, let’s go through the individual scripts and unpack the whole solution piece by piece. The following Python script is using ARM templates to provision Azure resource group and a small Citus database cluster (single coordinator node and two worker nodes).<\/p>\n
\r\nimport adal\r\nfrom timeit import default_timer as timer\r\nfrom sys import platform\r\nfrom msrestazure.azure_active_directory import AdalAuthentication\r\nfrom msrestazure.azure_cloud import AZURE_PUBLIC_CLOUD\r\nfrom azure.mgmt.resource.resources.models import (\r\n DeploymentMode,\r\n DeploymentProperties,\r\n Deployment,\r\n)\r\nfrom azure.mgmt.resource import ResourceManagementClient\r\nfrom pathlib import Path, PurePosixPath\r\nimport json\r\nimport os\r\n\r\narm_template_location = PurePosixPath(\r\n "\/home\/airflow\/airflow\/dags\/azure_citus_db_dag_deps\/azure_arm_templates\/pg_template.json"\r\n)\r\narm_template_params_location = PurePosixPath(\r\n "\/home\/airflow\/airflow\/dags\/azure_citus_db_dag_deps\/azure_arm_templates\/pg_template_params.json"\r\n)\r\n\r\nresource_group = "citus_svr_resource_group"\r\nexternal_IP = os.popen("curl -s ifconfig.me").readline()\r\ndeployment_name = "azure_citus_deployment"\r\n\r\nexternal_ip_params = {\r\n "name": "ClientIPAddress",\r\n "startIPAddress": external_IP,\r\n "endIPAddress": external_IP,\r\n}\r\nadmin_login = {"value": "citus"}\r\nadmin_passwd = {"value": "your_citus_password"}\r\n\r\ntemplate = arm_template_location\r\nparams = arm_template_params_location\r\n\r\nif template:\r\n with open(template, "r") as json_file_template:\r\n template = json.load(json_file_template)\r\n\r\nif params:\r\n with open(params, "r") as json_file_template:\r\n params = json.load(json_file_template)\r\n params = {k: v for k, v in params["parameters"].items()}\r\n params["firewallRules"]["value"]["rules"][1] = external_ip_params\r\n params["administratorLogin"] = admin_login\r\n params["administratorLoginPassword"] = admin_passwd\r\n resource_group_location = params["location"]["value"]\r\n\r\n\r\n# Tenant ID for your Azure Subscription\r\nTENANT_ID = "123a27e2-7777-4d30-9ca2-ce960d430ef8"\r\n\r\n# Your Service Principal App ID\r\nCLIENT = "ae277f4e-882d-4f03-a0b5-b69275046123"\r\n\r\n# Your Service Principal Password\r\nKEY = "e7bbf0ed-d461-48c7-aace-c6a4822a123e"\r\n\r\n# Your Azure Subscription ID\r\nsubscription_id = "1236a74c-dfd8-4b4d-b0b9-a355d2ec793e"\r\n\r\nLOGIN_ENDPOINT = AZURE_PUBLIC_CLOUD.endpoints.active_directory\r\nRESOURCE = AZURE_PUBLIC_CLOUD.endpoints.active_directory_resource_id\r\n\r\ncontext = adal.AuthenticationContext(LOGIN_ENDPOINT + "\/" + TENANT_ID)\r\ncredentials = AdalAuthentication(\r\n context.acquire_token_with_client_credentials, RESOURCE, CLIENT, KEY\r\n)\r\n\r\nclient = ResourceManagementClient(credentials, subscription_id)\r\n\r\n\r\ndef create_citus_instance(resource_group):\r\n\r\n deployment_properties = {\r\n "mode": DeploymentMode.incremental,\r\n "template": template,\r\n "parameters": params,\r\n }\r\n\r\n deployment_async_operation = client.deployments.create_or_update(\r\n resource_group, deployment_name, deployment_properties\r\n )\r\n\r\n deployment_async_operation.wait()\r\n\r\n\r\ndef main():\r\n print("\\nCreating Azure Resource Group...")\r\n start = timer()\r\n client.resource_groups.create_or_update(\r\n resource_group, {"location": resource_group_location}\r\n )\r\n print("Creating Citus Cluster...")\r\n create_citus_instance(resource_group)\r\n end = timer()\r\n time = round(end - start, 1)\r\n print("Total Elapsed Deployment Time = {t} seconds".format(\r\n t=format(time, ".2f")))\r\n\r\n\r\nif __name__ == "__main__":\r\n main()\r\n<\/pre>\nThis file, called provision_citus_cluster.py, will be called from our Airflow DAG using PythonOperator and is the first step of our pipeline. The instance creation process should take a few minutes to provision and once created we can continue on and generate Citus cluster schema with all the required objects.<\/p>\n
In order to populate Citus database with the flat files data we first need to create the desired schema (in this case called tpc_ds) and all the required objects. Even though this process will not be populating all tables (there are only thirteen flat files to be loaded), the script will create all TPC-DS benchmark objects using a simple SQL file containing all necessary DDL statements. The actual SQL file can be viewed and downloaded from my OneDrive folder so to keep this post concise, I will only include the Python script which parses the DDL SQL statements and executes those, creating all required tables. This operation (called from create_citus_schema.py file) will generate a second object in our DAG’s chain of tasks.<\/p>\n
\r\nfrom pathlib import Path, PurePosixPath\r\nfrom sys import platform\r\nfrom psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT\r\nimport psycopg2\r\nimport sys\r\n\r\nsql_schema = PurePosixPath(\r\n "\/home\/airflow\/airflow\/dags\/azure_citus_db_dag_deps\/sql\/tpc_ds_schema.sql"\r\n)\r\n\r\npg_args = {\r\n "user": "citus",\r\n "password": "your_citus_password",\r\n "host": "citussvrgroup-c.postgres.database.azure.com",\r\n "port": "5432",\r\n "database": "citus",\r\n "sslmode": "require",\r\n}\r\n\r\npg_schema_name = "tpc_ds"\r\n\r\n\r\ndef get_sql(sql_schema):\r\n """\r\n Source operation types from the 'create_sqlite_schema' SQL file.\r\n Each operation is denoted by the use of four dash characters\r\n and a corresponding table DDL statement and store them in a dictionary\r\n (referenced in the main() function).\r\n """\r\n table_name = []\r\n query_sql = []\r\n\r\n with open(sql_schema, "r") as f:\r\n for i in f:\r\n if i.startswith("----"):\r\n i = i.replace("----", "")\r\n table_name.append(i.rstrip("\\n"))\r\n\r\n temp_query_sql = []\r\n with open(sql_schema, "r") as f:\r\n for i in f:\r\n temp_query_sql.append(i)\r\n l = [i for i, s in enumerate(temp_query_sql) if "----" in s]\r\n l.append((len(temp_query_sql)))\r\n for first, second in zip(l, l[1:]):\r\n query_sql.append("".join(temp_query_sql[first:second]))\r\n sql = dict(zip(table_name, query_sql))\r\n return sql\r\n\r\n\r\ndef build_schema(sql_schema, pg_schema_name, **kwargs):\r\n connection = None\r\n try:\r\n connection = psycopg2.connect(**kwargs)\r\n cursor = connection.cursor()\r\n cursor.execute("SELECT version();")\r\n record = cursor.fetchone()\r\n if record:\r\n connection.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)\r\n cursor.execute(\r\n "CREATE SCHEMA IF NOT EXISTS {s};".format(\r\n s=pg_schema_name)\r\n )\r\n conn_params = connection.get_dsn_parameters()\r\n if conn_params["dbname"] == "citus":\r\n sql = get_sql(sql_schema)\r\n ddls = sql.values()\r\n for s in ddls:\r\n try:\r\n cursor.execute(s)\r\n connection.commit()\r\n except (Exception, psycopg2.DatabaseError) as error:\r\n print("Error while creating nominated database object", error)\r\n sys.exit(1)\r\n else:\r\n raise Exception("Failed to connect to citus database.")\r\n\r\n except (Exception, psycopg2.Error) as error:\r\n print("Could not connect to your PostgreSQL instance.", error)\r\n finally:\r\n if connection:\r\n cursor.close()\r\n connection.close()\r\n\r\n\r\ndef main():\r\n build_schema(sql_schema, pg_schema_name, **pg_args)\r\n\r\n\r\nif __name__ == "__main__":\r\n main()\r\n<\/pre>\nUp next is our third task which loads text files data into the newly created database. The following Python code (solution file is called load_data_into_citus.py) is responsible for taking each of the staged flat files and copying those into the corresponding tables (in a sequential order).<\/p>\n
\r\nimport psycopg2\r\nimport os\r\nimport sys\r\nfrom pathlib import Path, PurePosixPath\r\nfrom psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT\r\n\r\n\r\ntpc_ds_files = PurePosixPath(\r\n "\/home\/airflow\/airflow\/dags\/azure_citus_db_dag_deps\/tpc_ds_data\/"\r\n)\r\n\r\npg_schema_name = "tpc_ds"\r\nencoding = "iso-8859-1"\r\npg_args = {\r\n "user": "citus",\r\n "password": "your_citus_password",\r\n "host": "citussvrgroup-c.postgres.database.azure.com",\r\n "port": "5432",\r\n "database": "citus",\r\n "sslmode": "require",\r\n}\r\n\r\n\r\ndef load_data(tpc_ds_files, pg_schema_name, **kwargs):\r\n connection = None\r\n files = [f for f in os.listdir(tpc_ds_files) if f.endswith(".csv")]\r\n try:\r\n connection = psycopg2.connect(**kwargs)\r\n cursor = connection.cursor()\r\n cursor.execute("SELECT version();")\r\n record = cursor.fetchone()\r\n if record:\r\n connection.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)\r\n for file in files:\r\n schema_table_name = pg_schema_name + "." + file[:-4]\r\n with open(\r\n os.path.join(tpc_ds_files, file), "r", encoding=encoding\r\n ) as f:\r\n # next(f) # Skip the header row\r\n print(\r\n "Truncating table {table_name}...".format(\r\n table_name=schema_table_name\r\n )\r\n )\r\n cursor.execute(\r\n "TRUNCATE TABLE {table_name};".format(\r\n table_name=schema_table_name\r\n )\r\n )\r\n print(\r\n "Copying file {file_name} into {table_name} table...\\n".format(\r\n file_name=file, table_name=schema_table_name\r\n )\r\n )\r\n cursor.copy_from(f, schema_table_name, sep="|", null="")\r\n file_row_rounts = sum(\r\n 1\r\n for line in open(\r\n os.path.join(tpc_ds_files, file),\r\n encoding=encoding,\r\n newline="",\r\n )\r\n )\r\n cursor.execute(\r\n "SELECT COUNT(1) FROM {tbl}".format(\r\n tbl=schema_table_name)\r\n )\r\n record = cursor.fetchone()\r\n db_row_counts = record[0]\r\n\r\n if file_row_rounts != db_row_counts:\r\n raise Exception(\r\n "Table {tbl} failed to load correctly as record counts do not match: flat file: {ff_ct} vs database: {db_ct}.\\\r\n Please troubleshoot!".format(\r\n tbl=schema_table_name,\r\n ff_ct=file_row_rounts,\r\n db_ct=db_row_counts,\r\n )\r\n )\r\n\r\n except Exception as e:\r\n print(e)\r\n finally:\r\n if connection:\r\n cursor.close()\r\n connection.close()\r\n\r\n\r\ndef main():\r\n load_data(tpc_ds_files, pg_schema_name, **pg_args)\r\n\r\n\r\nif __name__ == "__main__":\r\n main()\r\n<\/pre>\nAs you may expect, each one of the aforementioned operations needs to be executed sequentially as there are clear dependencies across each of the steps outlined above i.e. database objects cannot be populated without the schema being created in the first place and likewise, database schema cannot be created without the database instance being present. However, when it comes to SQL queries execution, Citus was designed to crunch large volumes of data concurrently. Our next set of operation will calculate and generate four separate SQL results and stage those as CSV files inside the ‘queries_output’ folder. Please also note that this post does not look into Citus cluster performance and is not concerned with the queries execution times. We are not attempting to tune the way the data is distributed across the nodes or adjust any configuration parameters as the main purpose of this post is to highlight Airflow functionality and design patterns for the above scenarios.<\/p>\n
\r\nimport psycopg2\r\nimport os\r\nimport sys\r\nimport csv\r\nfrom pathlib import Path, PurePosixPath\r\n\r\n\r\nsql_queries_file = PurePosixPath("\/home\/airflow\/airflow\/dags\/azure_citus_db_dag_deps\/sql\/queries.sql")\r\nsql_output_files_dir = PurePosixPath(\r\n "\/home\/airflow\/airflow\/dags\/azure_citus_db_dag_deps\/queries_output\/"\r\n )\r\n\r\npg_schema_name = "tpc_ds"\r\nencoding = "iso-8859-1"\r\npg_args = {\r\n "user": "citus",\r\n "password": "your_citus_password",\r\n "host": "citussvrgroup-c.postgres.database.azure.com",\r\n "port": "5432",\r\n "database": "citus",\r\n "sslmode": "require",\r\n}\r\n\r\n\r\ndef get_sql(sql_queries_file):\r\n query_number = []\r\n query_sql = []\r\n\r\n with open(sql_queries_file, "r") as f:\r\n for i in f:\r\n if i.startswith("----"):\r\n i = i.replace("----", "")\r\n query_number.append(i.rstrip("\\n"))\r\n temp_query_sql = []\r\n with open(sql_queries_file, "r") as f:\r\n for i in f:\r\n temp_query_sql.append(i)\r\n l = [i for i, s in enumerate(temp_query_sql) if "----" in s]\r\n l.append((len(temp_query_sql)))\r\n for first, second in zip(l, l[1:]):\r\n query_sql.append("".join(temp_query_sql[first:second]))\r\n sql = dict(zip(query_number, query_sql))\r\n return sql\r\n\r\n\r\ndef main():\r\n query_sql = sql.get(param).rstrip()\r\n query_sql = query_sql[:-1] # remove last semicolon\r\n output_file_name = param.lower() + "_results.csv"\r\n try:\r\n connection = psycopg2.connect(**pg_args)\r\n cursor = connection.cursor()\r\n cursor.execute("SELECT version();")\r\n record = cursor.fetchone()\r\n if record:\r\n sql_for_file_output = "COPY ({0}) TO STDOUT WITH CSV DELIMITER '|';".format(query_sql)\r\n with open(os.path.join(sql_output_files_dir, output_file_name), "w") as output_file:\r\n cursor.copy_expert(sql_for_file_output, output_file)\r\n except Exception as e:\r\n print(e)\r\n finally:\r\n if connection:\r\n cursor.close()\r\n connection.close()\r\n\r\n\r\nif __name__ == "__main__":\r\n if len(sys.argv[1:]) == 1:\r\n sql = get_sql(sql_queries_file)\r\n param = sys.argv[1]\r\n query_numbers = [q for q in sql]\r\n if param not in query_numbers:\r\n raise ValueError(\r\n "Incorrect argument given. Choose from the following numbers: {q}".format(\r\n q=" or ".join(query_numbers)\r\n )\r\n )\r\n else:\r\n param = sys.argv[1]\r\n main()\r\n else:\r\n raise ValueError(\r\n "Too many arguments given. Looking for <query number> numerical value."\r\n )\r\n<\/pre>\nThe above code is executed using BashOperator rather than PythonOperator, as each query is run with a parameter passed to the executing script, indicating the SQL query number. Additionally, each query output (the result set) is staged as a CSV file on the local drive so that at the end of the pipeline execution we should end up with four flat files in the queries_output folder.<\/p>\n
Last but not least, we will initialize the ‘clean-up’ step, where Citus cluster resource group and all its corresponding services will be terminated to avoid incurring additional cost and send an email to a nominated email address, notifying administrator of the successful pipeline execution completion. The following Python code terminates and deletes all resources created in this process.<\/p>\n
\r\nimport adal\r\nfrom timeit import default_timer as timer\r\nfrom sys import platform\r\nfrom msrestazure.azure_active_directory import AdalAuthentication\r\nfrom msrestazure.azure_cloud import AZURE_PUBLIC_CLOUD\r\nfrom azure.mgmt.resource.resources.models import (\r\n DeploymentMode,\r\n DeploymentProperties,\r\n Deployment,\r\n)\r\nfrom azure.mgmt.resource import ResourceManagementClient\r\n\r\nresource_group = "citus_svr_resource_group"\r\n\r\n\r\n# Tenant ID for your Azure Subscription\r\nTENANT_ID = "123a27e2-7777-4d30-9ca2-ce960d430ef8"\r\n\r\n# Your Service Principal App ID\r\nCLIENT = "ae277f4e-882d-4f03-a0b5-b69275046123"\r\n\r\n# Your Service Principal Password\r\nKEY = "e7bbf0ed-d461-48c7-aace-c6a4822a123e"\r\n\r\n# Your Azure Subscription ID\r\nsubscription_id = "1236a74c-dfd8-4b4d-b0b9-a355d2ec793e"\r\n\r\nLOGIN_ENDPOINT = AZURE_PUBLIC_CLOUD.endpoints.active_directory\r\nRESOURCE = AZURE_PUBLIC_CLOUD.endpoints.active_directory_resource_id\r\n\r\ncontext = adal.AuthenticationContext(LOGIN_ENDPOINT + "\/" + TENANT_ID)\r\ncredentials = AdalAuthentication(\r\n context.acquire_token_with_client_credentials, RESOURCE, CLIENT, KEY\r\n)\r\n\r\nclient = ResourceManagementClient(credentials, subscription_id)\r\n\r\n\r\ndef main():\r\n client.resource_groups.delete(resource_group)\r\n\r\n\r\nif __name__ == "__main__":\r\n main()\r\n<\/pre>\n
There are many ways one can script out email notification dispatch using Python, however, Airflow provides out-of-the-box functionality in the form of EmailOperator. For this demonstration I decided to use my personal Gmail address which required a slight change to the airflow.cfg (Airflow configuration) file as well as creating a Google account App Password. More on the process of creating App Password can be found HERE<\/a>. The following changes to the Airflow configuration file are required (in addition to DAG-specyfic email task definition) to enable EmailOperator functionality with Gmail-specyfic email address.<\/p>\n
\r\nsmtp_host = smtp.gmail.com\r\nsmtp_user = your_gmail_email_address@gmail.com\r\nsmtp_password = your_app_password\r\nsmtp_port = 587\r\nsmtp_mail_from = Airflow\r\n<\/pre>\n
The final piece of the puzzle is the actual DAG file responsible for individual tasks definition as per my previous introductory post to Airflow HERE<\/a>. The following Python code is responsible for defining DAG arguments, individual tasks and the order they should be executed in. Notice that some Python tasks are executed using BashOperator instead of PythonOperator. That’s because those scripts are run with an argument passed to them using Python sys module argv list, denoting query number which the script is to run. Also, those tasks need to include full path to where the script is located. In contrast, PythonOperator only requires a Python callable to be included.<\/p>\n
\r\nfrom airflow import DAG\r\nfrom airflow.operators.bash_operator import BashOperator\r\nfrom airflow.operators.python_operator import PythonOperator\r\nfrom datetime import datetime, timedelta\r\nfrom airflow.operators.email_operator import EmailOperator\r\nfrom azure_citus_db_dag_deps import create_citus_schema\r\nfrom azure_citus_db_dag_deps import load_data_into_citus\r\nfrom azure_citus_db_dag_deps import provision_citus_cluster\r\nfrom azure_citus_db_dag_deps import destroy_citus_cluster\r\n\r\n\r\ndefault_args = {\r\n 'owner': 'Airflow',\r\n 'depends_on_past': False,\r\n 'start_date': datetime(2019, 6, 1),\r\n 'email': ['airflow@example.com'],\r\n 'email_on_failure': False,\r\n 'email_on_retry': False,\r\n 'retries': 0,\r\n # 'retry_delay': timedelta(minutes=5),\r\n # 'queue': 'bash_queue',\r\n # 'pool': 'backfill',\r\n # 'priority_weight': 10,\r\n # 'end_date': datetime(2016, 1, 1),\r\n}\r\n\r\nwith DAG('Azure_Citus_Database_Job', default_args=default_args, schedule_interval=None) as dag:\r\n provisioning_citus_cluster = PythonOperator(\r\n task_id="Provision_Citus_Cluster",\r\n python_callable=provision_citus_cluster.main)\r\n creating_citus_schema = PythonOperator(\r\n task_id="Create_DB_Schema", python_callable=create_citus_schema.main)\r\n loading_csv_data = PythonOperator(\r\n task_id="Load_Data", python_callable=load_data_into_citus.main)\r\n executing_sql_query_5 = BashOperator(\r\n task_id="Run_Query_5", bash_command='python ~\/airflow\/dags\/azure_citus_db_dag_deps\/query_data_in_citus.py Query5'\r\n )\r\n executing_sql_query_13 = BashOperator(\r\n task_id="Run_Query_13", bash_command='python ~\/airflow\/dags\/azure_citus_db_dag_deps\/query_data_in_citus.py Query13'\r\n )\r\n executing_sql_query_47 = BashOperator(\r\n task_id="Run_Query_47", bash_command='python ~\/airflow\/dags\/azure_citus_db_dag_deps\/query_data_in_citus.py Query47'\r\n )\r\n executing_sql_query_57 = BashOperator(\r\n task_id="Run_Query_57", bash_command='python ~\/airflow\/dags\/azure_citus_db_dag_deps\/query_data_in_citus.py Query57'\r\n )\r\n destroying_citus_cluster = PythonOperator(\r\n task_id="Destroy_Citus_Cluster", python_callable=destroy_citus_cluster.main)\r\n\r\n emailing_operator = EmailOperator(\r\n task_id="Email_Operator",\r\n to="your_gmail_email_address@gmail.com",\r\n subject="Airflow Message",\r\n html_content="""<h1>Congratulations! All your tasks are now completed.<\/h1>"""\r\n )\r\n\r\nprovisioning_citus_cluster >> creating_citus_schema >> loading_csv_data >> executing_sql_query_5 >> destroying_citus_cluster >> emailing_operator\r\nprovisioning_citus_cluster >> creating_citus_schema >> loading_csv_data >> executing_sql_query_13 >> destroying_citus_cluster >> emailing_operator\r\nprovisioning_citus_cluster >> creating_citus_schema >> loading_csv_data >> executing_sql_query_47 >> destroying_citus_cluster >> emailing_operator\r\nprovisioning_citus_cluster >> creating_citus_schema >> loading_csv_data >> executing_sql_query_57 >> destroying_citus_cluster >> emailing_operator\r\n<\/pre>\nNow that we have everything in place we can kick this DAG off and wait for the pipeline to go through its individual execution steps, concluding with the queries output files staged in the nominated folder as well as the email sent to the Gmail address we specified.<\/p>\n
When finished, Airflow can provide us with the Gantt chart breakdown on how long individual tasks took which helps with pinpointing slow-running processes and gives us a good overview of time consumed across each step.<\/p>\n
![\"\"](\"http:\/\/bicortex.com\/bicortex\/wp-content\/post_content\/\/2020\/04\/Airflow_DAG_Execution_Gantt_Chart.png\")
<\/a><\/p>\nAdditionally, we can now see the DAG tree view completion status (green = success, red = failure) for each run and each step, the four output files staged in the nominated directory with expected queries result sets persisted and finally, the email notification sent out on pipeline successful completion.<\/p>\n
![\"\"](\"http:\/\/bicortex.com\/bicortex\/wp-content\/post_content\/\/2020\/04\/Airflow_DAG_Tree_View.png\")
<\/a>![\"\"](\"http:\/\/bicortex.com\/bicortex\/wp-content\/post_content\/\/2020\/04\/Airflow_Job_Completion_Email.png\")
<\/a><\/p>\nConclusion<\/h3>\n
This concludes this short series outlining some of the core functionality of Airflow workflow management system. From the short time I spent with Airflow it seems like it fills the void for a robust, open-source, configuration-as-code platform which can be used across many different application, not just ETL\/ELT e.g. automating DevOps operations, machine learning pipelines, scheduler replacement etc. Its documentation is passable, it has a fairly mild learning curve, it’s fairly extensible and scalable, has a good management interface and its adoption rate, particularly in the startup community, is high as it is slowly becoming the de facto platform for programmatic workloads authoring\/scheduling. Data engineering field has been changing rapidly over the last decade, competition is rife and I’m not sure whether Airflow will be able to sustain its momentum in years to come. However, given its current rise to stardom, easily superseding tools such as Apache Oozie or Luigi, anyone in the field of Data Engineering should be at least familiar with it.<\/p>\n","protected":false},"excerpt":{"rendered":"
Airflow Sample Workflow Continued… Note: Part 1 can be found HERE and all files used in this tutorial can be downloaded from HERE. In the first part to this short series on Apache Airflow I outlined it its core architecture, installation process and created a sample example of a very rudimentary DAG (Directed Acyclic Graph). […]<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[61,72,56,5],"tags":[65,62,79,81,73,24,41,49],"class_list":["post-4027","post","type-post","status-publish","format-standard","hentry","category-cloud-computing","category-mpp-rdbms","category-programming","category-sql","tag-azure","tag-cloud-computing","tag-data-warehouse","tag-etl","tag-mpp-rdbms","tag-programming","tag-python","tag-sql"],"aioseo_notices":[],"jetpack_featured_media_url":"","_links":{"self":[{"href":"http:\/\/bicortex.com\/bicortex\/wp-json\/wp\/v2\/posts\/4027","targetHints":{"allow":["GET"]}}],"collection":[{"href":"http:\/\/bicortex.com\/bicortex\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"http:\/\/bicortex.com\/bicortex\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"http:\/\/bicortex.com\/bicortex\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"http:\/\/bicortex.com\/bicortex\/wp-json\/wp\/v2\/comments?post=4027"}],"version-history":[{"count":72,"href":"http:\/\/bicortex.com\/bicortex\/wp-json\/wp\/v2\/posts\/4027\/revisions"}],"predecessor-version":[{"id":4369,"href":"http:\/\/bicortex.com\/bicortex\/wp-json\/wp\/v2\/posts\/4027\/revisions\/4369"}],"wp:attachment":[{"href":"http:\/\/bicortex.com\/bicortex\/wp-json\/wp\/v2\/media?parent=4027"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"http:\/\/bicortex.com\/bicortex\/wp-json\/wp\/v2\/categories?post=4027"},{"taxonomy":"post_tag","embeddable":true,"href":"http:\/\/bicortex.com\/bicortex\/wp-json\/wp\/v2\/tags?post=4027"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}