Automating Workflows: How Apache Airflow can streamline Data Processes - Step by Step Guide

Apache Airflow is an open-source platform for developing, scheduling, and monitoring batch-oriented workflows. Airflow’s extensible Python framework enables you to build workflows connecting with virtually any technology. A web interface helps manage the state of your workflows. Airflow is deployable in many ways, varying from a single process on your machine to a distributed setup to support even the biggest workflows.
Airflow is used to author workflows as Directed Acyclic Graphs (DAGs) of tasks. The Airflow scheduler executes your tasks on an array of workers while following the specified dependencies. Rich CLI makes performing complex surgeries on DAGs a snap. The rich user interface makes it easy to visualize pipelines running in production, monitor progress, and troubleshoot issues when needed.

Requirements:

Source	Main Version (Dev)	Stable Version (2.10.0)
Python	3.8, 3.9, 3.10, 3.11, 3.12	3.8, 3.9, 3.10, 3.11, 3.12
Platform	AMD64/ARM64(*)	AMD64/ARM64(*)
Kubernetes	1.28, 1.29, 1.30, 1.31	1.27, 1.28, 1.29, 1.30
PostgreSQL	12, 13, 14, 15, 16	12, 13, 14, 15, 16
MySQL	8.0, 8.4, Innovation	8.0, 8.4, Innovation
SQLite	3.15.0+	3.15.0+

Workflows as Code:

The main characteristic of Airflow workflows is that all workflows are defined in Python code. Workflows as code serve several purposes.

1. Dynamic: Airflow pipelines are configured as Python code, allowing for dynamic pipeline generation.
2. Extensible: The Airflow framework contains operators to connect with numerous technologies. All Airflow components are extensible to easily adjust to your environment.
3. Flexible: Workflow parameterization is built-in leveraging the Jinja templating engine.
4. Elegant: Airflow pipelines are lean and explicit. Parameterizing your scripts is built into the core of Airflow using the powerful Jinja templating engine.
5. Scalable: Airflow has a modular architecture and uses a message queue to orchestrate an arbitrary number of workers

Implementation sources:

Apache Airflow can be implemented using the below methods.

1. PyPi
2. Docker Images
3. Docker Files
4. Helm Charts
5. Released sources
6. Managed Airflow Services.

The recommended method for implementing, running, and installing Apache Airflow is to use a combination of Docker image/Dockerfile and Docker compose file. It provides the capability of running Airflow components in isolation from other software running on the same physical or virtual machines with easy maintenance of dependencies.

Installing Pre-requisites: 

Install Docker Community Edition (CE) on your workstation. Depending on your OS, you may need to configure Docker to use at least 4.00 GB of memory for the Airflow containers to run properly.

Install Docker Compose v2.14.0 or newer on your workstation.

Implementation Steps:

We will create a Docker compose YAML file for installing and deploying the Apache Airflow Docker container. This docker-compose will have the following service definitions

 - Airflow Scheduler which monitors all tasks and DAGs, then triggers the task instances once their dependencies are complete.

 - Airflow Webserver that provides a UI at specified IP:PORT URL.

 - Airflow Worker that executes the tasks given by the scheduler.

 - Airflow Triggerer that runs an event loop for deferrable tasks.

 - Airflow Init as the initialization service.

 - Postgres as the Database.

 - Redis is the broker that forwards the messages from the scheduler to the worker.

For this Docker compose, we will have mounted directories so that their contents are synchronized between the machine and the container. Directories would be running on user-based ownership which is generally for the user Airflow. 

 - DAGs Directory: The DAG files are stored here.

 - Logs Directory: It contains logs from task execution and scheduler.

 - Config Directory: This has the local configuration file airflow_local_settings.py for custom implementation.

 - Plugins Directory: Used to add custom plugins.

The crucial step before starting the implementation is to create the above directories and set the correct permissions for the Airflow user, as the Docker container will be running via Airflow user. Docker compose has a lot of services listed in it, but it’s important to note the UID and GID of the Airflow user on the machine to make sure the Docker compose has the correct values. 

mkdir -p ./dags ./logs ./plugins ./config

echo -e "AIRFLOW_UID=$(id -u)" > .env

Now we run the Docker command to deploy the Airflow-Init service which initialize the Database and user login. The account created has the login airflow and the password airflow.

docker compose up airflow-init

Once this has been deployed, we need to clean up the environment for running all the services listed in the docker compose. 

Run the below command in the directory you downloaded the docker-compose YAML file.

Once removed, recreate the Docker compose using the guide and store it in the Directory. 

docker compose down --volumes --remove-orphans

Now we can start all services

docker compose up -d –build

The above implementation can be processed using a Dockerfile instead of a Docker Image which provides more customization using a requirements text file reference inside the Dockerfile. To do that, just Create a Dockerfile in the same directory where your docker-compose YAML file is stored. Place the requirements text file in the same directory as the Dockerfile and Docker compose YAML file. Run the command to deploy the containers using the Dockerfile instead of the image.

docker compose up -d –build

Once the cluster has started up, you can log in to the web interface and begin experimenting with DAGs.

Using Airflow:

Let’s look at the following snippet of code (DAG):

DAG Breakdown:

DAG Definition:

dag_id = dag_with_taskflow_api_v02

This is the unique identifier for the DAG.

default_args

Defines default parameters with retries (5 attempts) and delay between retries (5 minutes).

start_date = datetime(2024, 09, 04) 

The DAG can start running from this date. If the DAG is triggered after this date, any missed runs will be backfilled unless the catchup behavior is turned off.

Tasks Breakdown:

get_name()

Returns a dictionary containing a first name and last name: 

{'first_name': 'Smit', 'last_name': 'Shanischara'}.

get_age()

Returns a static integer, representing an age.

greet()

Prints a greeting message 

Task Execution Flow:

The DAG starts by executing get_name() and get_age() tasks in parallel since there is no dependency between them.

Once these two tasks are complete, their outputs are passed to the greet() task, which uses the extracted data to print a message.

Summarizing, This DAG demonstrates a simple workflow where data is "extracted" (name and age), combined in a task, and then "loaded" by printing a greeting message.

Airflow evaluates this script and executes the tasks at the set interval and in the defined order. The status of the DAG is visible in the web interface:

Just like the above DAG, we can have multiple DAGs running on a single Apache Airflow instance. 

Removing Airflow: 

To stop and delete containers, delete volumes with database data, and download images, run:

docker compose down --volumes --rmi all

Best Practices: 

 - Modular DAGs: Break large workflows into smaller, reusable DAGs. This promotes better maintainability and scalability.

 - Task Dependencies: Properly define task dependencies using set_upstream() and set_downstream() methods to ensure correct execution order.

 - Handle Failures: Use retries, on_failure_callback, and alerts to handle task failures gracefully.

 - Version Control: Keep your DAGs under version control (e.g., Git) for better collaboration and tracking changes.

 - Monitoring: Set up monitoring for failed tasks and performance bottlenecks using tools like Grafana or integrating with cloud monitoring solutions.

 - Security: Secure Airflow with role-based access control (RBAC) and authentication mechanisms.

 - Logging: Ensure logs are properly stored in a centralized system for debugging and compliance.

References:

https://airflow.apache.org/docs/apache-airflow/stable/installation/index.html#using-production-docker-images

https://github.com/apache/airflow

https://airflow.apache.org/docs/apache-airflow/2.2.5/

If you have any questions you can reach out our SharePoint Consulting team here.

Harnessing SharePoint OOTB Events with PnP Commands for Automated Permission Updates & Efficient Operations

Introduction

In SharePoint Online, automating operations in response to internal events — such as the creation of new site groups or granting permissions — can significantly streamline administrative tasks. This is particularly vital in environments with multiple associated sites where permissions need to be synchronized regularly. Here's how we can use SharePoint's OOTB events to improve operational efficiency and ensure timely updates across sites.

Challenges in Synchronizing Permissions

Consider a typical scenario in a SharePoint environment with one hub site and several associate sites. When a new user is added to a group in the hub site, the same user must be added across all associated sites. 

If we use the Logic App or Power Automate to sync the permission across all these sites it would require daily SharePoint group checks, leading to permissions synchronization delays.

Innovative Solution Using PnP Commands

To address this delay, we leveraged SharePoint PnP (Patterns and Practices) commands to trigger Flows or Azure Functions directly via HTTP when specific OOTB events occur. This method provides immediate event details such as usernames and group names, which can then be used to synchronize user permissions across sites instantly.

Setup and Configuration

We set up event receivers in SharePoint that respond to specific user group changes. Here are the PnP PowerShell commands used to configure these event receivers:

1. When a User is Added:

 Add-PnPEventReceiver -Name "GroupUserAdded" -Url $functionUrlToAddUser -EventReceiverType GroupUserAdded -Synchronization Asynchronous  

2. When a User is Removed:

 Add-PnPEventReceiver -Name "GroupUserDeleted" -Url $functionUrlToRemoveUser -EventReceiverType GroupUserDeleted -Synchronization Asynchronous  

These commands integrate two Azure Functions: the first triggers when a user is added to any SharePoint group on the Hub site, and the second when a user is removed. This setup ensures that any modifications to user groups are immediately reflected across all associated sites without the need for manual intervention.

Benefits and Capabilities

By leveraging OOTB events, we can perform a variety of operations that respond dynamically to changes within the SharePoint environment. These include but are not limited to:

• Adding, updating, or deleting items, lists, and sites
• Managing group and role definitions and assignments
• Handling file movements and attachment operations

Conclusion

Utilizing SharePoint's OOTB events with PnP commands to trigger automated actions eliminates delays inherent in daily synchronization tasks. This approach not only enhances operational efficiency but also improves user experience by ensuring that permissions and other configurations are consistently up-to-date. This strategy ensures more efficient management of SharePoint systems, minimizing delays and boosting overall operational effectiveness.

 If you have any questions you can reach out our SharePoint Consulting team here.

Resolving Power Automate Flow Trigger Issues when Migrating Items to a SharePoint List

Introduction:

Automation in SharePoint using Power Automate can significantly streamline processes and enhance productivity. However, certain scenarios, such as bulk item migrations, can lead to unintended behaviors. This blog post will share a recent problem encountered with an automated flow triggered by item creation in a SharePoint list. We will also walk you through the steps we took to resolve this issue effectively.

Problem:

We had set up an automated flow in Power Automate that trigger whenever a new item is created in a SharePoint list. Before migrating many items, we turned off this flow to prevent it from triggering for each migrated item. After completing the migration and reactivating the flow, we observed an unexpected behavior, the flow triggered for all the newly migrated items due to data stored in the browser cache. This resulted in uneven and unwanted executions of the flow.

Steps to the solution:

To avoid this issue in future migrations, we followed an approach that ensures the flow does not trigger unintentionally during bulk item migrations. Here are the steps:

Step 1: Export the Flow

Before starting the migration process, export the existing flow to your local machine. This step ensures that you have a backup of the flow configuration and can easily restore it later.

1. Open Power Automate.
2. Navigate to the flow which you want to export.
3. Select the flow and open it.
4. Select "Export", then click on the "Package (.zip)"
5. In the opened "Package details" section, Give the Name of the package.
6. Click on the exported .zip file and download it to your computer.
   
    

Step 2: Delete the Existing Flow

With the flow exported and safely stored, proceed to delete the existing flow. This step prevents any triggers from occurring during the migration process.

1. Navigate to the flow which you want to delete.
   
2. Click on “Delete” and it will ask for confirmation. 
   

Step 3: Migrate the Items

Now that the flow has been deleted, proceed with migrating the items to the SharePoint list. Since the flow is not present, it will not trigger during this process.

Step 4: Recreate the Flow by Importing the Zip File

After completing the migration, recreate the flow by importing the previously exported .zip file. This ensures that the flow is restored to its original configuration.

1. Navigate to the location where you want to Import.
   
2. Click on "Import" -> Import Package (Legacy).
   
3. Upload the .zip file that you exported earlier; it will take some time to complete the process.
4. In Review Package Content – click on the "Update" link from IMPORT SETUP.
5. It will open an "Import setup" panel, in the Setup field we need to select the "Create as new" option and then click on the Save button.
6. Click on the Import button. [Note: It will take some time to Import the file, so please do not navigate away]
7. Once the Import is completed then you will get a successfully imported message.
   
   

8. 

Step 5: Verify and Reactivate the Flow Once the flow is successfully imported, verify its configuration to ensure everything is set up correctly. Then, reactivate the flow.

1. Locate the newly imported flow.
   
2. Ensure the configuration is correct and click on "Turn on" to reactivate it. 
   
   

Conclusion:

By following these steps, you can effectively manage your automated flows in Power Automate and avoid unintended triggers during bulk item migrations. Exporting and deleting the flow before migration, and then recreating it afterward, ensures that your automation behaves as expected and maintains data integrity. This method provides a reliable solution to prevent the issues caused by browser cache and uneven flow triggering.

Implementing this approach will save you time and effort, ensuring a smooth migration process without disrupting your automated workflows.

If you have any questions you can reach out our SharePoint Consulting team here.