DEV Community: Dr. Malte Polley

Enhancing Your Matillion Python Component with a Custom Matillion Python Package Powered by Amazon Q Developer

Dr. Malte Polley — Tue, 15 Jul 2025 07:01:38 +0000

At MRH Trowe, we've integrated Matillion with Snowflake and AWS to streamline our ELT and ETL data integrations (Extract, Load, Transform). Matillion offers a low-code/no-code solution for creating these integrations, providing both SQL-related components (Transformation) for transformation pipelines and orchestration pipelines for data integration and automation (Extract and Load).

While Matillion comes with many pre-built components to connect with various sources, there are situations where Python components are necessary for more complex tasks or to implement feature which are not supported. This combination allows us to load data from different sources into Snowflake, leveraging its robust data platform capabilities.

Given Matillion's focus on low-code integration, I often wondered how to streamline Python development for use within Matillion. Considering that Matillion operates on Docker in AWS (e.g., with Fargate), I decided it was best to take ownership of the container by creating a custom Matillion Helper Package for Python to ease integration. For this, I utilized Amazon Q Developer.

Amazon Q can be leveraged with VS Code and it's server version using a dedicated free extension. You can use your Builders profile or leverage SSO (Amazon Identity Center) to streamline IAM-related tasks for Amazon Q Developer. After installing this extension, you're provided with a chat interface that facilitates an agentic development lifecycle.

Creating a Custom PyPi Package with Amazon Q

I set out to create my first Python package as a companion to the Matillion Dockerfile with the following structure:

matillion-helper/
├── matillion_helper/
│   ├── __init__.py
│   └── helper.py
├── setup.py
├── pyproject.toml
├── README.md
├── Dockerfile
└── .dockerignore

To generate this structure, I prompted Amazon Q: “Create the needed minimal setup for PyPi package to be installed into a Docker container. Name the package matillion_helper.” With Claude Sonnet 3.7. this took only a matter of seconds.

Use Case: Consolidated Logging

Once the package was set up, I focused on creating a module for consolidated logging. This module would log messages from any function up to the main execution, making logging consistent across the package and all Matillion projects. The prompt "Create a logging_config.py file for my matillion_helper package with the following features:

Functions to configure logging with console and file output
Support for log rotation with configurable max size and backup count
Environment variable support for log directory and log level
A get_logger convenience function
Proper error handling for file permissions
Type hints for all functions"

As a result I got this code.


import logging
import logging.handlers
import os
import sys
from pathlib import Path
from typing import Any, Optional

def configure_logging(
    logger_name: Optional[str] = None,
    log_level: int = logging.INFO,
    log_file: Optional[str] = None,
    log_format: str = "[%(asctime)s] [%(levelname)s] [%(name)s] - %(message)s",
    date_format: str = "%Y-%m-%d %H:%M:%S",
    console_output: bool = True,
    max_bytes: int = 10485760,  # 10MB
    backup_count: int = 5,
    propagate: bool = False,
) -> logging.Logger:
    """Configure logging for a module.

    Args:
        logger_name: Name of the logger (defaults to the calling module name)
        log_level: Logging level (default: INFO)
        log_file: Log file name (default: based on logger_name)
        log_format: Format string for log messages
        date_format: Format string for timestamps
        console_output: Whether to output logs to console
        file_output: Whether to output logs to file
        max_bytes: Maximum size of log file before rotation
        backup_count: Number of backup log files to keep
        propagate: Whether to propagate logs to parent loggers

    Returns:
        logging.Logger: Configured logger instance
    """
    # If no logger name is provided, use the calling module's name
    if logger_name is None:
        import inspect

        frame = inspect.stack()[1]
        module = inspect.getmodule(frame[0])
        logger_name = module.__name__ if module else "matillion_helper"

    # Get or create the logger
    logger = logging.getLogger(logger_name)

    # Clear any existing handlers to avoid duplicate logs
    if logger.hasHandlers():
        logger.handlers.clear()

    # Set the log level
    logger.setLevel(log_level)
    logger.propagate = propagate

    # Create formatter
    formatter = logging.Formatter(log_format, date_format)

    # Add console handler if requested
    if console_output:
        console_handler = logging.StreamHandler(sys.stdout)
        console_handler.setFormatter(formatter)
        console_handler.setLevel(log_level)
        logger.addHandler(console_handler)

    return logger

def get_logger(
    name: Optional[str] = None,
    log_level: Optional[int] = None,
    **kwargs: Any,
) -> logging.Logger:
    """Get a configured logger.

    This is a convenience function for getting a logger with the default configuration.

    Args:
        name: Logger name (defaults to the calling module name)
        log_level: Override default log level
        **kwargs: Additional arguments to pass to configure_logging

    Returns:
        logging.Logger: Configured logger instance
    """
    # Get environment variable for log level, defaulting to INFO
    env_log_level = os.environ.get("MATILLION_LOG_LEVEL", "INFO")

    # Map string log levels to logging constants
    log_level_map = {
        "DEBUG": logging.DEBUG,
        "INFO": logging.INFO,
        "WARNING": logging.WARNING,
        "ERROR": logging.ERROR,
        "CRITICAL": logging.CRITICAL,
    }

    # Use provided log_level, or get from environment, or default to INFO
    effective_log_level = log_level or log_level_map.get(env_log_level, logging.INFO)

    return configure_logging(logger_name=name, log_level=effective_log_level, **kwargs)

This implementation ensured that the logging mechanism was centralized and not dependent on individual developers' preferences.

Use Case: Interaction with AWS API

Another regular task we conduct with Matillion involves IAM context switches. This setup allows us to eliminate static credentials. To achieve this, we implemented the ability to make STS (Security Token Service) AssumeRole calls. The Matillion instance needs permissions from both the source and target AWS accounts to facilitate cross-account access.

Using a prompt like “Create a function to make STS AssumeRole possible in a generic way,” I developed a module that utilizes the existing logging_config.py for consistent logging throughout the implementation.

import boto3
from botocore.exceptions import ClientError, ParamValidationError
from typing import Dict, Optional, Any
from matillion_helper.logging_config import get_logger

# Get configured logger for this module
logger = get_logger(__name__)

def assume_role(
    role_arn: str,
    session_name: str = "AssumedRoleSession",
    duration_seconds: int = 3600,
    external_id: Optional[str] = None,
    region_name: Optional[str] = None,
    sts_client: Optional[Any] = None,
) -> Dict[str, Any]:
    """Assume an AWS IAM role and return temporary credentials.

    This function allows for assuming an IAM role and obtaining temporary credentials
    that can be used to make AWS API calls with the permissions of the assumed role.

    Args:
        role_arn (str): The Amazon Resource Name (ARN) of the role to assume.
        session_name (str, optional): An identifier for the assumed role session.
            Defaults to "AssumedRoleSession".
        duration_seconds (int, optional): The duration, in seconds, of the role session.
            Defaults to 3600 (1 hour).
        external_id (str, optional): A unique identifier that might be required when
            assuming a role in another account. Defaults to None.
        region_name (str, optional): The AWS region to connect to. Defaults to None,
            which uses the default region from the AWS configuration.
        sts_client (boto3.client, optional): An existing STS client to use.
            If not provided, a new client will be created.

    Returns:
        Dict[str, Any]: A dictionary containing the temporary credentials and session information.
        The dictionary includes:
            - AccessKeyId: The access key ID for the temporary credentials
            - SecretAccessKey: The secret access key for the temporary credentials
            - SessionToken: The session token for the temporary credentials
            - Expiration: When the temporary credentials expire

    Raises:
        ClientError: If AWS returns an error during role assumption.
        ParamValidationError: If the parameters provided are invalid.
        Exception: Any other exception that occurs during role assumption.
    """
    try:
        # Create STS client if not provided
        if sts_client is None:
            sts_client = boto3.client("sts", region_name=region_name)

        # Prepare assume role parameters
        assume_role_params = {
            "RoleArn": role_arn,
            "RoleSessionName": session_name,
            "DurationSeconds": duration_seconds,
        }

        # Add external ID if provided
        if external_id:
            assume_role_params["ExternalId"] = external_id

        # Assume the role
        response = sts_client.assume_role(**assume_role_params)

        # Extract credentials from the response
        credentials = response["Credentials"]

        logger.info(f"Successfully assumed role: {role_arn}")

        return credentials

    except ClientError as e:
        logger.exception(f"Failed to assume role due to AWS error: {e}")
        raise
    except ParamValidationError as e:
        logger.exception(f"Invalid parameters for assume_role: {e}")
        raise
    except Exception as e:
        logger.exception(f"Unexpected error assuming role: {e}")
        raise

Use Case: Refreshing Power BI Semantic Models

Another useful feature we developed is the ability to kick off dataset refreshes for Power BI from Matillion. This change transforms a scheduled dependency between Matillion and Power BI semantic models into an event-based approach, resulting in more consistent behavior and fewer error sources. However, configuration in Power BI Admin Settings is required to allow API interaction, and there are restrictions on refreshes—limited to eight per day. If you meet the prerequisites, you leverage event based refreshes for all power bi workspaces and semantic models, by inniting the service principal and assigning contributor rights.

Using a prompt like “Create a Python function to update Power BI datasets in a workspace.”, I implemented a function that interacts with the Power BI API to refresh datasets, ensuring that we can maintain updated reports directly from our data integration workflows.


import requests
import json
import msal
from matillion_helper.logging_config import get_logger

# Get configured logger for this module
logger = get_logger(__name__)

def update_powerbi_datasets(workspace_id, client_id, client_secret, tenant_id):
    """
    Update Power BI datasets in a specified workspace.

    Args:
        workspace_id (str): The ID of the Power BI workspace
        client_id (str): Azure AD application client ID
        client_secret (str): Azure AD application client secret
        tenant_id (str): Azure AD tenant ID

    Returns:
        dict: Result of the refresh operation
    """
    # Get access token using MSAL
    authority = f"https://login.microsoftonline.com/{tenant_id}"
    app = msal.ConfidentialClientApplication(
        client_id=client_id,
        client_credential=client_secret,
        authority=authority
    )

    # Acquire token for Power BI service
    scopes = ["https://analysis.windows.net/powerbi/api/.default"]
    result = app.acquire_token_for_client(scopes=scopes)

    if "access_token" not in result:
        logger.exception(f"Error getting token: {result.get('error')}")
        logger.exception (f"Error description: {result.get('error_description')}")
        return None

    access_token = result["access_token"]

    # Get datasets in the workspace
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json"
    }

    datasets_url = f"https://api.powerbi.com/v1.0/myorg/groups/{workspace_id}/datasets"
    datasets_response = requests.get(datasets_url, headers=headers)

    if datasets_response.status_code != 200:
        logger.exception (f"Error getting datasets: {datasets_response.text}")
        return None

    datasets = datasets_response.json()["value"]
    refresh_results = {}

    # Refresh each dataset
    for dataset in datasets:
        dataset_id = dataset["id"]
        dataset_name = dataset["name"]

        refresh_url = f"https://api.powerbi.com/v1.0/myorg/groups/{workspace_id}/datasets/{dataset_id}/refreshes"
        refresh_response = requests.post(refresh_url, headers=headers)

        if refresh_response.status_code == 202:
            refresh_results[dataset_name] = "Refresh triggered successfully"
        else:
            refresh_results[dataset_name] = f"Error: {refresh_response.text}"

    return refresh_results

# Example usage
if __name__ == "__main__":
    # Replace these with your actual value
    client_id="client_id"
    client_secret="client_secret"
    tenant_id="tenant_id"
    workspace_id="workspace_id"

    results = update_powerbi_datasets(workspace_id, client_id, client_secret, tenant_id)
    logger.info(results)

Summary

By utilizing Amazon Q and creating a custom Matillion Python package, we gained much-needed flexibility in our development process. While this approach requires initial setup and containerization efforts, the benefits include a more integrated CI/CD process and the ability to manage Python scripts effectively within Matillion.

As of today, the current Copilot from Matillion (MAIA) is not helping you to create and maintain Python scripts. But by introducing Amazon Q Developer you can overcome this hurdle of hight code implementation by procideng meanigful buildings blocks and your data enginneers to leverage pure but predefined Python code.

Happy (vibed) Conding! :-)

Implementing Automated End-to-End Testing: Leveraging Your CI/CD with Your Cloud Development Kit (CDK) App

Dr. Malte Polley — Sat, 15 Mar 2025 07:52:29 +0000

I recently embarked on a new project focused on our input management. I’m working for a German insurance broker that aims to modernize the handling of physical letters. Unfortunately, in the German insurance sector, we cannot utilize APIs or similar technical implementations. Instead, our account managers deal with letters by reading them, extracting information, and finally inputting all relevant details into our CRM.

The volume of physical letters I’m referring to is approximately 8,500 pages per month. Therefore, ensuring the quality of our event-based architecture is crucial. In this article, I will demonstrate how we implemented an end-to-end (E2E) testing approach within our CI/CD pipeline using the Cloud Development Kit (CDK) to create the infrastructure.

The Architecture

I won’t spend too much time discussing the architecture itself, as I might write another blog post about this use case in the future. However, I will provide the minimal amount of information necessary to give you a clue. Essentially, Figure 1 illustrates everything. We have an on-premise scanner that forwards all scans to the Simple Email Service (SES). Each letter is treated as an email for SES. SES stores all emails as MIME objects within S3, and a Lambda function retrieves the PDF attachment, subsequently triggering AWS Step Functions to process the PDF. Using AWS Textract, we obtain the OCR results, and we utilize Amazon Bedrock to conduct intelligent document processing (IDP). From this IDP, we extract specific information to create a new filename (I know, old school!). Finally, a Lambda function pushes the processed PDF into an SMB file share.

Implementing a Test Strategy: Demystifying E2E Tests

I won’t reiterate all the blog posts about designing test cases, but I will highlight two fundamental aspects. First, we need to establish a test hierarchy. Unit tests form the base layer, integration tests are the mid-layer, while E2E tests represent the top layer in every test implementation case. Unit tests evaluate whether a specific input produces the expected output from a function. Integration tests assess whether a component can create and handle the results from a second component. E2E tests evaluate whether all components of a system work together as intended. While hopefully everyone is using unit tests for quality assurance of Lambdas and other code components, integration tests and E2E tests are considerably more complex and costly.

Let’s take a step back and analyze what the described architecture is doing to clarify the complexity of E2E testing. First, we need to process an email. Next, we process the PDF, and finally, we push the processed PDF file into a specific folder within our cloud environment. From this abstraction, our E2E tests need to simulate sending an email to mock the scanner. Secondly, we need to track the events generated by SES. This can be tricky if you want to capture the specific event and the Lambda function that processes the MIME object. Instead, I utilized my AWS Step Function. Listing the executions of this Step Function seemed like a good idea. Finally, I could create a lookup into my SMB file share to search for my test case PDF.

Prerequisites for This Approach

You might already be considering the assumptions we need to meet, right? If your CI/CD pipeline is not running within a network segment that can route to the SMB file share, you’re out of luck. Therefore, network integration can be crucial. Additionally, it’s essential to consider IAM permissions. We use the CDK bootstrap command to create all relevant components for the deployments. However, this collection of resources does not permit listing and describing resources unrelated to the CDK. Thus, part of your application needs to be an IAM role that can be assumed by your CI/CD server. Consequently, you will need to modify the CDK bootstrap collection to allow you to assume your new IAM role for testing purposes.

Finally, since the CDK supports dynamic resource naming, which you should utilize, we cannot rely on deterministic naming schemas. The solution is to expose the email, step function, and IAM role for testing through CloudFormation Outputs. Each output will have a human-readable and constant export name, allowing us to retrieve dynamic resource names.

Lastly, we need to verify our email and domain entities within SES to utilize both. For emails, you need to click a link, and for domains, you must add records to your domain service, whether it’s Route 53 or another provider.

Discussion Around This Approach

Let’s discuss the previous points. First, the scanner depicted in the figure is not utilized in this E2E testing scenario. Fair point. My question is: does the effort to implement the scanner itself genuinely justify the evaluation of the AWS architecture? The scanner might be located separately from your environment. Ultimately, you are correct; this is not entirely E2E, but we are quite close.

Another point I would like to mention is that this E2E test uses excellent test cases. This means we use PDFs that can be effectively evaluated and processed by Textract and Bedrock, ensuring that I know how the final file will look. From my perspective, you could criticize this as well, but the essence of E2E testing is not about quality assessment in terms of high volume. It’s about configuring the system correctly to achieve predictable results. The overall performance of the system is monitored in production.

I'm also hearing you say that certain parts of the architecture are not fully automated in the initial deployment, such as the SES entities. You are entirely correct. Within SES, we create independent resources like the rule to forward incoming emails to S3 and Lambda, as well as dependent resources like domain and email entities, which need to be approved or created and retrieved, such as the SMTP credentials. That said, there is manual effort involved, but it is a one-time task. As I mentioned earlier, moving towards E2E testing requires more time and manual intervention compared to unit testing.

Implementing the Script

Let’s dive into the scripts. First, here’s the IAM Role and a sample for the CloudFormation output:

from aws_cdk import aws_iam as iam, CfnOutput
from constructs import Construct

class IAMRoleStack(Stack):
    """Create the IAM deployment."""

    def __init__(
        self,
        scope: Construct,
        construct_id: str,
        **kwargs,
    ) -> None:
        """Create the IAM deployment.

        Args:
            scope (Construct): CDK App scope
        """
        role = iam.Role(
            self,
            id="TestIAMRole",
            assumed_by=iam.ServicePrincipal("ec2.amazonaws.com"),
            description="IAM Role for E2E testing",
        )

        role.add_to_policy(
            iam.PolicyStatement(
                actions=[
                    "states:ListExecutions",
                    "states:DescribeExecution",
                ],
                resources=["*"],
            )
        )

        CfnOutput(
            self.scope,
            id="RoleArn",
            value=role.role_arn,
            export_name="E2eTestIAMRole",
            description="ARN of the created IAM Role",
        )

Next, here’s the bash implementation for our CI/CD pipeline, in my case using Azure DevOps but it can be any CI/CD tool:

export AWS_DEFAULT_REGION=${{ parameters.AwsRegion }}
KST=$( aws sts assume-role --endpoint-url https://sts.${{ parameters.AwsRegion }}.amazonaws.com --role-arn arn:aws:iam::${{ parameters.AccountID }}:role/${{ parameters.CDKDeploymentRole }}-${{ parameters.AccountID }}-${{ parameters.AwsRegion }} --role-session-name $(Build.SourceVersion) --duration-seconds 3600)
export AWS_ACCESS_KEY_ID=$(echo $KST | jq -r .Credentials.AccessKeyId)
export AWS_SECRET_ACCESS_KEY=$(echo $KST | jq -r .Credentials.SecretAccessKey)
export AWS_SESSION_TOKEN=$(echo $KST | jq -r .Credentials.SessionToken)
export EMAIL_PROCESSOR=$(aws cloudformation describe-stacks --stack-name mrht-developer-posteingang-stack --query "Stacks[0].Outputs[?ExportName=='ExportEmailProcessor'].OutputValue" --output text)
export STATE_MACHINE_ARN=$(aws cloudformation describe-stacks --stack-name mrht-developer-posteingang-stack --query "Stacks[0].Outputs[?ExportName=='ExportStateMachine'].OutputValue" --output text)
export IAM_ROLE_ARN=$(aws cloudformation describe-stacks --stack-name mrht-developer-posteingang-stack --query "Stacks[0].Outputs[?ExportName=='E2eTestIAMRole'].OutputValue" --output text)

KST=$( aws sts assume-role --endpoint-url https://sts.${{ parameters.AwsRegion }}.amazonaws.com --role-arn $IAM_ROLE_ARN --role-session-name $(Build.SourceVersion) --duration-seconds 3600)
unset AWS_ACCESS_KEY_ID AWS_SECRET_ACCESS_KEY AWS_SESSION_TOKEN
export AWS_ACCESS_KEY_ID=$(echo $KST | jq -r .Credentials.AccessKeyId)
export AWS_SECRET_ACCESS_KEY=$(echo $KST | jq -r .Credentials.SecretAccessKey)
export AWS_SESSION_TOKEN=$(echo $KST | jq -r .Credentials.SessionToken)
python3 src/e2e_test/e2e_test.py

The first aws sts assume-role command helps us to retrieve the CloudFormation outputs. The required permissions for this are part of the CDK bootstrap collection. The second aws sts assume-role command enables us to initiate the testing.

Now, let’s take a look at the e2e_test.py file:

import os
import time
import logging
import boto3
from botocore.exceptions import ClientError, ParamValidationError
import smbclient
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.application import MIMEApplication

logging.basicConfig(
    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

class SMBClient:
    """Manage an SMB connection to access shared files."""

    def __init__(self, username: str, password: str, servername: str):
        """Initialize the SMBClient with connection details.

        Args:
            username (str): The username for the SMB connection.
            password (str): The password for the SMB connection.
            servername (str): The hostname or IP address of the SMB server.
        """
        self.username = username
        self.password = password
        self.port = 445
        self.host_name = servername

    def connect_smb(self, smb_client: smbclient) -> None:
        """Connect to the SMB server.

        Args:
            smb_client (smbclient): The SMB client instance.
        """
        logger.info(f"Trying to connect to {self.host_name} as {self.username}.")
        try:
            smb_client.register_session(
                server=self.host_name, username=self.username, password=self.password
            )
            logger.info("Session established successfully.")
        except Exception as e:
            logger.exception(f"Failed to connect to SMB share: {e}")
            raise
        return True

    def list_directory(self, path: str, smb_client: smbclient) -> None:
        """List the contents of an SMB directory.

        Args:
            path (str): The path to the directory.
            smb_client (smbclient): The SMB client instance.

        Returns:
            List[str]: A list of the directory contents, or an empty list if an error occurs.
        """
        entries = []
        try:
            for entry in smb_client.listdir(path):
                logger.info(f"Found entry: {entry}")
                entries.append(entry)
            return entries
        except Exception as e:
            logger.exception(f"Error listing directory: {e}")
            return []

def send_email(
    sender: str,
    recipient: str,
    subject: str,
    body_path: str,
    attachment_path: str,
    smtp_server: str,
    smtp_port: int,
    smtp_username: str,
    smtp_password: str,
):
    """Send an email with a PDF attachment using SMTP.

    Args:
        sender (str): The email address of the sender.
        recipient (str): The email address of the recipient.
        subject (str): The subject of the email.
        body_path (str): The path to the text file containing the body of the email.
        attachment_path (str): The path to the PDF file to be attached.
        smtp_server (str): The SMTP server address.
        smtp_port (int): The SMTP server port.
        smtp_username (str): The SMTP username.
        smtp_password (str): The SMTP password.
    """
    msg = MIMEMultipart()
    msg["Subject"] = subject
    msg["From"] = sender
    msg["To"] = recipient

    with open(body_path, "r", encoding="utf-8") as text_file:
        text = MIMEText(text_file.read())
        msg.attach(text)

    with open(attachment_path, "rb") as pdf_file:
        attachment = MIMEApplication(pdf_file.read(), _subtype="pdf")
        attachment.add_header(
            "Content-Disposition", "attachment", filename="attachment.pdf"
        )
        msg.attach(attachment)

    try:
        with smtplib.SMTP(smtp_server, smtp_port) as s:
            s.starttls()
            s.login(smtp_username, smtp_password)
            s.send_message(msg)
            logger.info("Email sent successfully.")
    except Exception as e:
        logger.error(f"Error sending email: {str(e)}")
        raise


def main():
    """Run E2E test."""
    try:
        state_machine_arn = os.getenv("STATE_MACHINE_ARN")
        stage = os.getenv("STAGE")
        smb_user = os.getenv("SMB_USERNAME")
        smb_password = os.getenv("SMB_PASSWORD")
        smb_host_ip = os.getenv("SMB_SERVER")
        smb_path = f"{smb_host_ip}\\folder$\\folder\\"
        smtp_server = os.getenv("SMTP_SERVER")
        smtp_port = int(os.getenv("SMTP_PORT"))
        smtp_username = os.getenv("SMTP_USERNAME")
        smtp_password = os.getenv("SMTP_PASSWORD")
    except KeyError as e:
        logger.exception(e)
        raise

    try:
        sender = "scan2mail@example.com"
        recipient = "scan2mail@example.com"
        subject = "Test Email with Attachment"
        body_path = "./src/e2e_test/email_body.txt"
        attachment_path = "./src/e2e_test/test.pdf"

        send_email(
            sender=sender,
            recipient=recipient,
            subject=subject,
            body_path=body_path,
            attachment_path=attachment_path,
            smtp_server=smtp_server,
            smtp_port=smtp_port,
            smtp_username=smtp_username,
            smtp_password=smtp_password,
        )

        time.sleep(15)

        stepfunctions_client = boto3.client("stepfunctions")

        list_executions_response = stepfunctions_client.list_executions(
            stateMachineArn=state_machine_arn, maxResults=1, statusFilter="RUNNING"
        )

        if list_executions_response["executions"]:
            execution_arn = list_executions_response["executions"][0]["executionArn"]
            logger.info(f"Found execution: {execution_arn}")

            while True:
                response = stepfunctions_client.describe_execution(
                    executionArn=execution_arn
                )
                status = response["status"]

                if status != "RUNNING":
                    logger.info(f"Step Function execution status: {status}")
                    break

                logger.info(
                    "Step Function is still running. Checking again in 20 seconds..."
                )
                time.sleep(20)
        else:
            logger.warning("No running execution found.")
            raise

    except (ClientError, ParamValidationError) as e:
        logger.error(f"An error occurred: {str(e)}")
        raise

    counter = 0
    try:
        smb = SMBClient(
            username=smb_user,
            password=smb_password,
            servername=smb_host_ip,
        )
        smb.connect_smb(smbclient)
        response = smb.list_directory(smb_path, smbclient)
        desired_contract_id = "__102865946013"
        for file_name in response:
            if desired_contract_id in file_name:
                logger.info(f"Found the ID '{desired_contract_id}' in the directory contents.")
                counter += 1
    except Exception as e:
        logger.exception(e)
        raise

    if counter == 0:
        raise ValueError("Target ID not found")

    logger.info("Finished E2E test")

if __name__ == "__main__":
    try:
        main()
    except Exception as e:
        logger.error(f"An error occurred: {str(e)}")
        raise

This script is executed as the last step of my CI/CD pipeline after a successful CDK deployment. First, we gather all relevant variables from the CI/CD pipeline environment. Obviously, you could also use AWS services like the Simple Systems Manager (SSM) Parameter Store or AWS Secrets Manager. With that, we grab a test.pdf from the repository, as well as a test body for the email. If you leverage multiple cases, it’s probably a better idea to implement a test case store in S3.
Using the credentials and the test case, we can create an email that will be sent to our desired email address. In this case, the sender and recipient are the same. After the email has been successfully sent, we wait for a while and then proceed to list and describe our Step Function as shown in Figure 1. If the Step Function completes successfully, the script lists the SMB target directory and searches for our desired contract ID from the test case PDF. Finally, we clean up the test from the SMB directory.

Final Thoughts

The conception and implementation of E2E tests undoubtedly require significant effort, and I completely agree that this is a relatively simple example. However, automation helps ensure quality and saves a substantial amount of time.

The manual process would look something like this: sending an email using your favorite client, hoping you don’t overlook the correct test case, waiting for the file to appear in the SMB file store, and checking whether the contract ID matches your test case. This manual approach would certainly take longer than the discussed automated method, which takes about 1 minute and 30 seconds.

What do you think? Did I get things right? What is your opinion? I’m happy to discuss further and answer any questions you may have.

Happy Coding :-)!

Cloud Development Kit (CDK) Networking Deployment – Introduction and Overcoming a Defect in AWS Resource Access Management

Dr. Malte Polley — Sun, 02 Feb 2025 11:10:54 +0000

Introduction: The Need for Network Operations Centers in AWS

When I first began creating AWS infrastructure, cross-account deployments were not common practice. However, over the past seven years, this landscape has changed significantly. Two primary reasons for distributing workloads across multiple accounts are the concepts of segregation of duties and the ability to exceed common AWS account API and resource limits.

During my time at AWS, I had the opportunity to design and implement Networking Operations Centers (NoCs). The purpose of a NoC is to centralize certain aspects of network design within an organization’s AWS cloud infrastructure. This approach is relatively straightforward to implement.

First, you should utilize the Transit Gateway as a centralized router for your networking needs. Next, create private Virtual Private Clouds (VPCs) and connect them to the Transit Gateway. You can set these up locally in each AWS account or distribute the VPCs using AWS Resource Access Manager (RAM) from your networking account to others. To manage IP addresses effectively and avoid overlapping CIDRs in a dynamic environment, consider using the Amazon VPC IP Manager, which allows for centralized CIDR assignment from a designated IP pool.

Additionally, you should deny the creation of local Internet Gateways in each account through Service Control Policies (SCPs). Implement a security VPC with a firewall of your choice, designed with a gateway load balancer. Finally, establish centralized egress and ingress Internet access VPCs, or integrate this functionality into your security VPC. For more detailed information, I highly recommend reading "Building a Scalable and Secure Multi-VPC AWS Network Infrastructure."

Simplified Network Diagram

To visualize our approach at MRH-Trowe, consider the following simplified network diagram. All resources are created in the networking account and shared across various business units. By adopting this method, we prevent changes to the underlying network infrastructure by business units, establish a robust compliance foundation, and simplify processes for application developers, who typically do not focus on network infrastructure.

Automating Your Network Deployment: Leveraging the Cloud Development Kit (CDK)

Automation is crucial for creating and maintaining AWS workloads, especially when you need to build and tear down resources daily. The Cloud Development Kit (CDK) provides a programmatic approach to infrastructure management via AWS CloudFormation. While I won’t delve deeply into CDK basics, I encourage you to explore this tool if you haven't already—it's a game changer.

To create a layer-based approach for our NoC, we can structure our project with the following layers:

IP Address Manager (ipam_stack.py): This manages the IP pool we want to control.
Security VPC and Transit Gateway (security_vpc_stack.py): This layer handles security and routing.
Gateway Load Balancer Based Firewall (firewall_stack.py): This ensures robust security measures.

After these foundational layers, we can add multiple VPCs with automated routable entries and Transit Gateway attachments.

network-operation-center/
|-app.py
|-stacks/
|--ipam_stack.py
|--security_vpc_stack.py
|--firewall_stack.py
|--vpc1_stack.py
…
|--vpc_n_stack.py

While the first three stacks are somewhat specialized, we can leverage a more generic Python class for our VPCs, allowing for easier sharing later on. Below is an example of how to define a VPC construct tailored for MRH-Trowe's network:

from aws_cdk import (
    Aws,
    Tags,
    CfnTag,
    Duration,
    RemovalPolicy,
    aws_ec2 as ec2,
    aws_iam as iam,
    aws_ram as ram,
)
from stacks.security_vpc_stack import SecurityVPCStack
from constructs import Construct

class VPCDefinition(Construct):
    """Create a general-purpose VPC customized construct."""

    def __init__(
        self,
        scope: Construct,
        id: str,
        vpc_net_mask: int,
        vpc_name: str,
        default_cidr: str,
        transit_gateway_id: str,
        network_account_id: str,
        security_vpc_id: str = None,
        gwlb_param: str = None,
        ipv4_ipam_pool_id: str = None,
        max_azs: int = 2,
        amount_of_natgateways: int = 0,
        tgw_attach_cidr_mask: int = 28,
        public_service_cidr_mask: int = None,
        private_service_cidr_mask: int = None,
        isolated_service_cidr_mask: int = None,
        local_cidr: str = None,
        gwlb_service_name: str = None,
        tgw_attachement: bool = True,
    ) -> None:
        """Initialize CDK construct class."""
        super().__init__(scope, id)
        self.subnet_config = []
        self.tgw_subnet_ids = []
        self.tgw_route_table_ids = []
        self.tgw_azs = []
        self.id_collection = []
        self.vpc_net_mask = vpc_net_mask
        self.default_cidr = default_cidr
        self.transit_gateway_id = transit_gateway_id
        self.network_account_id = network_account_id
        self.ipv4_ipam_pool_id = ipv4_ipam_pool_id
        self.tgw_attach_cidr_mask = tgw_attach_cidr_mask
        self.public_service_cidr_mask = public_service_cidr_mask
        self.private_service_cidr_mask = private_service_cidr_mask
        self.isolated_service_cidr_mask = isolated_service_cidr_mask
        self.vpc_name = vpc_name
        self.max_azs = max_azs
        self.amount_of_natgateways = amount_of_natgateways
        self.local_cidr = local_cidr
        self.security_vpc_id = security_vpc_id
        self.tgw_attachement = tgw_attachement

        # Create a subnet set for the TGW
        if self.tgw_attach_cidr_mask == 28:
            self.subnet_config.append(
                {
                    "cidrMask": tgw_attach_cidr_mask,
                    "name": f"tgw-attachment-{self.vpc_name}",
                    "subnetType": ec2.SubnetType.PRIVATE_ISOLATED,
                    "MapPublicIpOnLaunch": False,
                }
            )
        # Additional subnet configurations can be added here as needed

        # Request a CIDR from IPAM
        ip_assignment = ec2.IpAddresses.aws_ipam_allocation(
            ipv4_ipam_pool_id=ipv4_ipam_pool_id,
            ipv4_netmask_length=self.vpc_net_mask,
        )
        # Create the actual VPC
        self.vpc = ec2.Vpc(
            self,
            id="VPC",
            ip_addresses=ip_assignment,
            max_azs=self.max_azs,
            nat_gateways=self.amount_of_natgateways,
            subnet_configuration=self.subnet_config,
            vpc_name=vpc_name,
            restrict_default_security_group=True,
        )
        # Tagging the VPC
        tags = {
            "Shared": "True",
            "SourceAccountId": self.network_account_id,
            "SourceAccountName": "Networking",
            "AWS Region": Aws.REGION,
        }
        for key, value in tags.items():
            Tags.of(self.vpc).add(key, value)

        # Collect subnet and route table IDs for automatic route table adoption
        if self.tgw_attach_cidr_mask == 28:
            selection_tgw_attach = self.vpc.select_subnets(
                subnet_group_name=f"tgw-attachment-{self.vpc_name}"
            )
            for i in selection_tgw_attach.subnets:
                self.id_collection.append(
                    {
                        "RouteTableId": i.route_table.route_table_id,
                        "Az": i.availability_zone,
                    }
                )
                self.tgw_subnet_ids.append(i.subnet_id)

        # Create Transit Gateway attachments and route table propagations
        transit_gateway_attachment = ec2.CfnTransitGatewayAttachment(
            self,
            id="TransitGatewayAttachment",
            subnet_ids=self.tgw_subnet_ids,
            transit_gateway_id=self.transit_gateway_id,
            vpc_id=self.vpc.vpc_id,
            tags=[CfnTag(key="Name", value=f"{vpc_name}-tgw-attachment")],
            options={"ApplianceModeSupport": "enable"},
        )
        # Create the TGW route table assocaition
        ec2.CfnTransitGatewayRouteTableAssociation(
            self,
            id="SpokeRouteTableAssociation",
            transit_gateway_attachment_id=transit_gateway_attachment.ref,
            transit_gateway_route_table_id=self.tgw_spoke_route_table_id,
        )
        # Create route for the each subnet towards the security VPC with 0.0.0.0/0
        for idx, i in enumerate(self.id_collection):
            ec2.CfnRoute(
                self,
                id=f"Route{idx}",
                route_table_id=i["RouteTableId"],
                destination_cidr_block=default_cidr,
                transit_gateway_id=transit_gateway_id,
            ).node.add_dependency(transit_gateway_attachment)

These construct lifts a lot of work, but can be used easily later as the following examples show:

from constructs import Construct
from stacks.ipam_stack import IPAMStack
from stacks.security_vpc_stack import SecurityVPCStack
from stacks.network_creation import VPCDefinition
from aws_cdk import (
    Stack,
)

class SomeVPC(Stack):
    """Create the actual deployment in each AWS account.

    Args:
        Stage (Stage): cdk Class stage
    """

    def __init__(
        self, scope: Construct, construct_id: str, **kwargs
    ) -> None:
        """Intitialise CDK stack class."""
        super().__init__(scope, construct_id, **kwargs)
        """Create the actual CloudFormation stack."""
        self.vpc_name = "my-vpc"
        self.network_account_id = "12345678910"
        self.param_name_ipam_pool = "ipam-id"
        self.param_name_tgw_id = "tgw-id"
        self.default_cidr = "0.0.0.0/0"
        self.security_vpc_name = "my-sec-vpc"

        self.ipv4_ipam_pool_id = IPAMStack.return_pool_id(
            stack=self, pool_name=self.param_name_ipam_pool
        )
        self.transit_gateway_id = SecurityVPCStack.return_tgw_id(
            stack=self, param_name_tgw=self.param_name_tgw_id
        )

        vpc = VPCDefinition(
            self,
            id=self.vpc_name,
            transit_gateway_id=self.transit_gateway_id,
            network_account_id=self.network_account_id,
            ipv4_ipam_pool_id=self.ipv4_ipam_pool_id,
            default_cidr=self.default_cidr,
            vpc_name=self.vpc_name,
            vpc_net_mask=26,
            public_service_cidr_mask=28,
        )

       ram.CfnResourceShare(
            id="CfnResourceVPCShare",
            name=name_resource_share,
            allow_external_principals=False,
            principals="12345678911",
            resource_arns=["subnet_arn_1,subnet_arn_2, … subnet_arn_n"],
        )

As you can see, we just need to pass some parameters and a new VPCs is born. With ram.CfnResourceShare(), we share the subnets via AWS Resource Access Manager with a certain target account. By sharing the subnets, the whole VPC will be shared, including route tables, routes etc.

Using the Network Resources: The Power of Tags

When utilizing network resources, the CDK offers a convenient method called ec2.Vpc.from_lookup(). This allows you to access all relevant information using the VPC ID. However, a significant issue arises: AWS Resource Access Manager does not share the tags created by the CDK for the network infrastructure components. This hinders the lookup functionality essential for effective resource management.

aws-cdk:subnet-type:isolated and aws-cdk:subnet-name:private-service-mrht-teamviewer-vpc. These tags are crucial for the lookup functionality.

To mitigate this defect, we can leverage Custom Resources—AWS Lambda-backed components within our CDK app that gather tags in the networking account and replicate them to the destination account. The Lambda function requires a role in your networking account to describe route tables and subnets. We need to remove tags which starts with 'aws', otherwise, you will see this error: An error occurred (InvalidParameterValue) when calling the CreateTags operation: Value ( aws:cloudformation:stack-name ) for parameter key is invalid.

import json
import boto3
import os
import logging
import urllib3
from botocore.exceptions import ClientError, ParamValidationError

log_level = os.environ.get("LOG_LEVEL", "INFO")
logging.root.setLevel(logging.getLevelName(log_level))
logger = logging.getLogger(__name__)
http = urllib3.PoolManager()

ec2_client = boto3.client("ec2")
sts_client = boto3.client("sts")

def send(
    http,
    event,
    context,
    response_status,
    response_data,
    physical_resource_id=None,
    no_echo=False,
    reason="-",
):
    """Build CustomResource.

    Args:
        http (urllib3.PoolManager): object for put requests
        event (dict): Lambda event dict
        context (object): Lambda object
        response_status (string): result of this custom resource
        response_data (dict): additional data from this custom resource
        reason (string): error result of this custom resource
        physical_resource_id (string, optional): CloudFormation physical resource id. Defaults to None.
        no_echo (bool, optional): Echo mode activation. Defaults to False.

    Returns:
        No returns
    """
    response_url = event["ResponseURL"]
    logger.info(response_url)
    response_body = {}
    response_body["Status"] = response_status
    response_body["Reason"] = reason
    response_body["PhysicalResourceId"] = (
        physical_resource_id or context.log_stream_name
    )
    response_body["StackId"] = event["StackId"]
    response_body["RequestId"] = event["RequestId"]
    response_body["LogicalResourceId"] = event["LogicalResourceId"]
    response_body["NoEcho"] = no_echo
    response_body["Data"] = response_data

    json_response_body = json.dumps(response_body)
    logger.info("Response body:\n" + json_response_body)
    headers = {"content-type": "", "content-length": str(len(json_response_body))}

    try:
        response = http.request(
            "PUT",
            response_url,
            body=json_response_body.encode("utf-8"),
            headers=headers,
        )
        logger.info("Status code: " + response.reason)
        return True
    except Exception as e:
        logger.exception("send(..) failed executing requests.put(..): " + str(e))
        return False

def assume_role(role_arn: str, sts_client: boto3):
    """Assume IAM role in different Account.

    Args:
        sts_client (boto3, optional): boto3 object for STS. Defaults to sec_man_client.
        role_arn (str): SFTP Connection class. Defaults to SFTP.

    Raises:
        KeyError: KeyError. Missing environment variable.
        ClientError and ParamValidationError: ClientError. Boto3 Issue.

    Return:
        credentials: dict with tokens
    """
    try:
        response = sts_client.assume_role(
            RoleArn=role_arn, RoleSessionName="SyncSharedNetworkTags"
        )
    except (KeyError, ClientError, ParamValidationError) as e:
        raise e

    return response

def clean_tags(tags: list[dict]):
    """Remove tags which starts with 'aws'. An error occurred (InvalidParameterValue) when calling the CreateTags operation: 
    Value ( aws:cloudformation:stack-name ) for parameter key is invalid. Tag keys starting with 'aws:' are reserved for internal use.

    Params:
        tags (list[dict]):  tags gathered from describe calls

    Returns:
        tags: array with tags not starting with 'aws'
    """
    return [t for t in tags if not t['Key'].startswith('aws:')]

def main(event, context, sts_client=sts_client, source_ec2_client=ec2_client, http=http):
    """Create, delete and update custom actions.

    Params:
        sts_client: boto3 object for STS
        ec2_client: boto3 object for EC2
        http: http object
        event: Lambda event object
        context: Lambda context object
    """
    logger.info("Starting Network tag management ...")
    logger.info(event)
    logger.info(context)

    try:
        vpc_id = os.environ['VPC_ID']
        role_arn = os.environ['TARGET_IAM_ROLE_ARN']
    except KeyError as e:
        logger.exception(e)
        send(
            http=http,
            event=event,
            context=context,
            response_status="FAILED",
            response_data={"Response": str(e)},
            reason=str(e),
        )
        raise

    if event["RequestType"] == "Create" or event["RequestType"] == "Update":
        try:
            logger.info("Getting foreign credentials ...")
            target_credentials = assume_role(role_arn=role_arn, sts_client=sts_client)
            target_access_key = target_credentials["Credentials"]["AccessKeyId"]
            target_secret_access_key = target_credentials["Credentials"]["SecretAccessKey"]
            target_sessions_token = target_credentials["Credentials"]["SessionToken"]
        except (KeyError, ClientError, ParamValidationError) as e:
            logger.exception(e)
            send(
                http=http,
                event=event,
                context=context,
                response_status="FAILED",
                response_data={"Response": str(e)},
                reason=str(e),
            )
            raise

        logger.info("Creating foreign EC2 client ...")
        target_ec2_client = boto3.client(
            'ec2', 
            aws_access_key_id=target_access_key,
            aws_secret_access_key=target_secret_access_key,
            aws_session_token=target_sessions_token
        )
        logger.info("Done ...")

        try:
            logger.info("Describing local VPC tags ...")
            vpc_tags = source_ec2_client.describe_vpcs(VpcIds=[vpc_id])['Vpcs'][0]['Tags']
            logger.info("Copying local tags to shared VPC in foreign account ...")
            target_ec2_client.create_tags(Resources=[vpc_id], Tags=clean_tags(vpc_tags))
            logger.info("Done ...")

            logger.info("Describing local Subnet tags ...")
            subnets = source_ec2_client.describe_subnets(Filters=[{'Name': 'vpc-id', 'Values': [vpc_id]}])['Subnets']
            for subnet in subnets:
                subnet_id = subnet['SubnetId']
                subnet_tags = clean_tags(subnet['Tags'])
                logger.info("Copying local tags to shared Subnets in foreign account ...")
                logger.info(subnet_id)
                logger.info(subnet_tags)
                target_ec2_client.create_tags(Resources=[subnet_id], Tags=subnet_tags)
                logger.info(f"Done for subnet {subnet_id} ...")

            logger.info("Describing local Route Tables tags ...")
            route_tables = source_ec2_client.describe_route_tables(Filters=[{'Name': 'vpc-id', 'Values': [vpc_id]}])['RouteTables']
            logger.info(route_tables)
            for route_table in route_tables:
                route_table_id = route_table['RouteTableId']
                route_table_tags = clean_tags(route_table['Tags'])
                logger.info("Copying local tags to shared Route Tables with in foreign account ...")
                logger.info(route_table_id)
                logger.info(route_table_tags)
                if len(route_table_tags) > 0:
                    target_ec2_client.create_tags(Resources=[route_table_id], Tags=route_table_tags)
                else:
                    logging.info(f"Found Route Table {route_table_id} without tags ...")
                logger.info(f"Done for route table {route_table_id} ...")
        except (KeyError, ClientError, TypeError) as e:
            logger.exception(e)
            send(
                http=http,
                event=event,
                context=context,
                response_status="FAILED",
                response_data={"Response": str(e)},
                reason=str(e),
            )
            raise
        logger.info(f"Finished Network tag management on VPC {vpc_id} ...")
        send(
            http=http,
            event=event,
            context=context,
            response_status="SUCCESS",
            response_data={"Response": f"Finished Network tag management on VPC {vpc_id}"},
        )
        return True

    if event["RequestType"] == "Delete":
        logger.info(f"Deleting process in progress")
        send(
            http=http,
            event=event,
            context=context,
            response_status="SUCCESS",
            response_data={"Response": "Deleting process in progress"},
        )
        return True

Conclusion: A Secure and Scalable Network

In summary, this blog post illustrates how MRH-Trowe successfully created a robust AWS network infrastructure using a CDK-based approach. While the transition to a programmatic model may feel unusual for networking professionals accustomed to scripting or tools like Ansible, the benefits of a segregated duties approach are undeniable.

By investing time and effort into this project, we have established a secure and scalable network that empowers non-network-related teams to utilize existing resources without concern.

Happy coding!

Driving AWS Fargate to the Edge: Matillion Hybrid Agents and Python Pandas

Dr. Malte Polley — Sun, 22 Dec 2024 13:09:19 +0000

One of the standout services within the AWS ecosystem is the Elastic Container Service (ECS) with the Fargate launch type. This service is often referred to as “serverless” because it requires minimal operational setup to launch. However, while Fargate offers ease of use, it lacks one of the crucial benefits of AWS Lambda: automatic scaling. Regardless, Matillion leverages Fargate to create a cost-effective way to integrate your AWS environment with its services.

The Scalability Challenge with Big Data

While Matillion provides a straightforward integration, challenges can arise when discussing scalability and the movement of big data. For instance, if your data pipelines require processing gigabytes of data, this can be managed in a relatively static way using Matillion components. However, if you need to implement Python logic using Pandas, the situation becomes more complex.

You might be wondering: how do you handle gigabytes of data with a Fargate runner that has only 2 vCPUs and 4GB of memory? Yes, this can be tricky, especially if you need to wrangle the data during the transfer, as it must be loaded into memory.

Leveraging Python Pandas on Big Data with Fargate

A key issue to understand is that Python Pandas transforms data in memory. This design choice allows for high-speed processing, but it also introduces the risk of an “OutOfMemoryError,” which can lead to the termination of your Fargate task. If this happens, Matillion will not be able to track the state of your pipeline, as the executing task disappeared.

Your first reaction might be to increase the task size permanently. However, this could lead to a significant increase in your AWS bill. For instance, running production agents with Matillion’s suggested configuration (2 vCPUs, 4GB RAM, 20GB Storage) costs about $144.16 for a 24-hour period over 30 days. In contrast, scaling up to 16 vCPUs and 128GB RAM could result in costs of $1,724.28. Since not all pipelines require such extensive resources, introducing static scaling is not advisable. Instead, dynamic scaling could be an option, similar to how Snowflake resources can be adjusted using ALTER WAREHOUSE SQLs. However, managing dynamic scaling outside of your Matillion pipeline is essential to avoid losing tracking information. Having dynamic scaling from within a pipeline could be a valuable feature request for Matillion.

Implementing Chunking for Large Data Files

An easier implementation for your Pandas workloads can be achieved by using the sophisticated methods provided by the Pandas package itself. Since Pandas operates in memory, why not explore whether it has an option for chunking large data files into smaller parts? If so, can we apply transformation logic to each chunk? Does reading bytes introduce resilience against encoding issues and allow for generic data type identification in each chunk? Can we upload each chunk to the target storage system and free memory afterward? Finally, if we encounter errors, are we able to track them chunk by chunk?

The question of how to chunk a large DataFrame into smaller ones, such as converting a large CSV into multiple smaller Parquet files, is not new. Fortunately, today we have a chunksize parameter in methods like:

pandas.read_csv()
pandas.read_excel()
pandas.read_json()
pandas.read_sql()
pandas.read_hdf()

The authors of Pandas advise that chunking works well when the operation requires zero or minimal coordination between chunks. For more complicated workflows, other libraries might be a better choice.

Example of Chunking Implementation

Here’s a simple implementation of chunking:

import pandas as pd

chunk_size = 1000
chunk_number = 0

for chunk in pd.read_csv('large_file.csv', chunksize=chunk_size):
    chunk.to_parquet(f'chunk_{chunk_number}.parquet')
    chunk_number += 1

It’s important to note that the total number of rows is only meaningful if you know how many columns you have. Both dimensions drive the size of the DataFrame.

Applying Transformation Logic to Each Chunk

The answer to whether we can apply transformation logic for each chunk is a resounding yes. While chunking the data, we can apply transformations to each segment. However, we might encounter slight discrepancies in how Pandas interprets column data types, especially with CSV and JSON files where quotation marks and delimiters are crucial for resilient pipelines.

Here’s an example of applying a transformation to each chunk:

import pandas as pd

# Define a transformation function
def transform(df):
    df['numeric_column'] = df['numeric_column'].apply(lambda x: x * 1.1)
    return df

chunk_size = 1000
chunk_number = 0

for chunk in pd.read_csv('large_file.csv', chunksize=chunk_size):
    transformed_chunk = transform(chunk)
    transformed_chunk.to_parquet(f'chunk_{chunk_number}.parquet')
    chunk_number += 1

Handling Encoding Issues

To get a robust treatment of encodning errors, reading bytes is the better approach. However, it also has its drawbacks. For example, if we read the entire file at once, we risk hitting an OutOfMemory error:

import pandas as pd

# Define a transformation function
def transform(df):
    df['numeric_column'] = df['numeric_column'].apply(lambda x: x * 1.1)
    return df

file_path = 'large_file.csv'
chunk_size = 1000
chunk_number = 0

with open(file_path, mode="rb") as fd:
    for chunk in pd.read_csv(io.StringIO(fd.read().decode("latin_1")), chunksize=chunk_size):
        # Apply the transformation to each chunk
        transformed_chunk = transform(chunk)

        # Write the transformed chunk to a Parquet file
        transformed_chunk.to_parquet(f'chunk_{chunk_number}.parquet')
        chunk_number += 1

As a result, our Fargate Task will be killed by AWS ECS or the process with the Matillion Hybrid Agent will result in an Error 137, as chunking won’t take effect. Therefore, we need to adjust our code to read in chunks:

import pandas as pd
# Define a transformation function
def transform(df):
    df['numeric_column'] = df['numeric_column'].apply(lambda x: x * 1.1)
    return df

file_path = 'large_file.csv'
chunk_size = 10000
chunk_number = 0

with open (file_path, mode="rb") as fd:
    for chunk in pd.read_csv(
        fd,
        low_memory=False,
        chunksize=chunk_size,
        encoding="latin_1"
    ):
        # Apply the transformation to each chunk
        transformed_chunk = transform(chunk)

        # Write the transformed chunk to a Parquet file
        transformed_chunk.to_parquet(f'chunk_{chunk_number}.parquet')
        chunk_number += 1

This method ensures that encoding is applied in a chunk-wise manner, preventing memory issues. The parameter low_memory=False will be used to ensure data type consistency.

Uploading Chunks to Target Storage and Freeing Memory

To free memory, after a chunk has been created and uploading it to a target storage system, such as Amazon S3, here’s how you can do it:

import pandas as pd
import logging
import gc

# Define a transformation function
def transform(df):
    df['numeric_column'] = df['numeric_column'].apply(lambda x: x * 1.1)
    return df

file_path = 'large_file.csv'
s3_bucket = 'your_s3_bucket'
prefix_name = 'my_prefix'
file_name = 'large_file.csv'
access_key = 'your_access_key'
secret_access_key = 'your_secret_access_key'
sessions_token = 'your_session_token'

chunk_size = 10000
chunk_number = 0

with open (file_path, mode="rb") as fd:
    # Statt die gesamte Datei auf einmal zu lesen, verwenden wir einen iterativen Ansatz
    for chunk in pd.read_csv(
        fd,
        low_memory=False,
        chunksize=chunk_size,
        encoding="latin_1"
    ):
        # Apply the transformation to each chunk
        transformed_chunk = transform(chunk)

        # Define the Parquet file path
        parquet_file = f"s3://{s3_bucket}/{prefix_name}/part_{chunk_number}_{file_name.replace('.csv', '.snappy.parquet')}"

        # Write the transformed chunk to a Parquet file
        transformed_chunk.to_parquet(
            parquet_file,
            engine="auto",
            compression="snappy",
            storage_options={
                "key": access_key,
                "secret": secret_access_key,
                "token": sessions_token,
            },
        )

        logging.info(f"Uploaded chunk of {file_name} to {parquet_file}")
        chunk_number += 1
        del chunk
        gc.collect()

The parameter storage_options helps us to make of the boto3 implmentation of Pandas, while del chunk and gc.collect() free the memory.

Tracking Errors During Processing

Currently, tracking errors on a chunk-by-chunk basis can be challenging. However, you can leverage the on_bad_lines parameter introduced in Pandas v1.4 to receive warnings for lines that cannot be processed. With it's callback feature, you can write great log files to keep track on error understanding. This feature allows you to handle errors more effectively, although it may not be compatible with all configurations. This also the case, when using low_memory=False. In our usecase we are using on_bad_lines='warn' to parse aftwards the logs from Matillion.

Final Thoughts

The biggest drawback highlighted in this blog post is the lack of robust logging for errors. However, despite these challenges, using Matillion with AWS Fargate allows for cost-effective solutions without the need for complex clusters. By implementing chunking and careful memory management, you can avoid memory overflow errors and streamline the processing of large datasets.

As AWS, Matillion, and Snowflake continue to evolve, leveraging these tools together can optimize your data workflows, reduce execution time, and enhance overall efficiency. Remember, in the cloud, effective resource utilization is key to maximizing your investment.

Happy Coding :-)!

AWS Architectural Diagrams on a Commit Base: Using AWS PDK Diagram Plugin with Python

Dr. Malte Polley — Mon, 25 Nov 2024 09:05:30 +0000

One of the common challenges in providing sufficient and understandable information about AWS infrastructures is the creation of architectural diagrams. The process of creating these diagrams often involves certain assumptions about the audience's technical level and their familiarity with the diagram's content. This list is not meant to be exhaustive, but it highlights the complexities involved.

Creating these diagrams can be quite tedious and time-consuming. Additionally, AWS environments can change rapidly, which increases the workload if you want to keep everything up to date.

Approaches to Infrastructure Deployments with Visuals

Classic Approach

The classic approach involves creating the diagram upfront. This method is particularly useful when negotiating with stakeholders from IT security, customers, or data protection teams. While the diagram remains accurate until the first release date, its relevance tends to diminish over time.

Database Approach

To address this, other approaches focus on creating a database of the current AWS infrastructure. AWS itself provides solutions for this, but there are also AWS partners, like Lucid with their Lucid Spark product, which allows you to gather resources. While this approach keeps pace with the dynamic nature of AWS environments, the collection process is often not purposeful, as it typically gathers all resources indiscriminately. Consequently, a post-hoc collection and rearrangement process is necessary.

Visual Creation

A third approach involves visually creating the infrastructure and compiling the corresponding artifacts. This method is especially useful for beginners, as it allows them to interact with AWS indirectly and with a low barrier to entry. However, challenges arise when attempting to implement CI/CD processes or deploy multiple environments simultaneously.

Diagrams During or After Deployment

The fourth approach focuses on creating diagrams during or after deployment. This method aims to generate a resource representation based on your Terraform or CloudFormation scripts. While this approach keeps the diagrams aligned with the current architectural design, it lacks interactivity, which can be a significant drawback. The resource selection must be meaningful to foster a proper understanding of architecture.

Diagrams as Companions of the Cloud Development Kit (CDK)

The Cloud Development Kit (CDK) offers a powerful toolkit for creating complex and dynamic deployments in various ways. The CDK utilizes CloudFormation in combination with AWS services like S3, ECR, and Lambda, creating a dynamic abstraction that aids in deployment creation. In relation to the four approaches mentioned earlier, using the CDK aligns closely with the fourth approach.

One of the great features of the CDK is that it automatically builds a class tree of resources by default. This resource tree provides an opportunity to create a visual representation of the stack. Another advantage is the vibrant supporting community, where many talented individuals contribute new features daily. Finally, with support for native programming languages, I have yet to encounter a deployment task that could not be implemented with the CDK.

I recently found an open-source project that has the potential to improve the fourth approach significantly. The parent project is called cdk-graph.

Features of cdk-graph

Synthesizes a serialized graph (nodes and edges) from CDK source code.
Provides a runtime interface for interacting with the graph (in-memory database-like graph store).
Provides a plugin framework for additional tooling to utilize and extend the graph . The goal of this framework is to enable bespoke tooling to be built without having to first traverse the CDK Tree and Metadata to build a graph. Projects like cdk-dia generate a bespoke in-memory graph that is then utilized to generate diagrams. By standardizing on the graph interface necessary to build complex tooling, we can more rapidly build new tooling that focuses on its core value.

Implementing AWS PDK with Python

Let's create the CDK sample app:

XYZ@123:~/dev-to$ mkdir -p sample-app
XYZ@123:~/dev-to$ cd sample-app/
XYZ@123:~/dev-to/sample-app$ cdk init --language=python

This creates a simple CDK repository with a single SQS queue. I then added an example to my sample_app_stack.py. You can take another example if you like: AWS CDK Examples.

To run everything accordingly, we need to install these packages:

XYZ@123:~/dev-to/sample-app$source .venv/bin/activate
(.venv) XYZ@123:~/dev-to/sample-app$ pip install -r requirements.txt
(.venv) XYZ@123:~/dev-to/sample-app$ pip install aws-pdk asyncio

Now let's take a look at app.py, as this file needs to be changed a bit:

#!/usr/bin/env python3
from aws_pdk.cdk_graph import CdkGraph, FilterPreset, Filters, FilterValue
from aws_pdk.cdk_graph_plugin_diagram import CdkGraphDiagramPlugin
import asyncio
import aws_cdk as cdk
from sample_app.sample_app_stack import SampleAppStack

async def main():
    """Orchestration of CDK creation process. This function is needed to make things work in an asynchronous way."""
    app = cdk.App()
    SampleAppStack(
        app,
        "SampleRepoStack"
    )
    graph = CdkGraph(
        app,
        plugins=[
            CdkGraphDiagramPlugin(
                diagrams=[
                    {
                        "name": "diagram - 1 - filter compact",
                        "title": "Diagram SampleRepoStack",
                        "filterPlan": {
                            "preset": FilterPreset.COMPACT,
                        },
                    },
                    {
                        "name": "diagram - 2 - no filter",
                        "title": "Diagram SampleRepoStack",
                        "filterPlan": {
                            "preset": FilterPreset.NONE,
                        },
                    },
                    {
                        "name": "diagram - 3 - focus on API Gateway",
                        "title": "Diagram SampleRepoStack",
                        "filterPlan": {
                            "filters": [
                                {
                                    "graph": Filters.include_cfn_type([
                                        FilterValue(regex=r"AWS::ApiGateway::*")
                                    ]),
                                },
                            ],
                        },
                    }
                ]
            ),
        ],
    )
    app.synth()
    graph.report()

if __name__ == "__main__":
    """We call the orchestration function via asyncio lib to make things work in an asynchronous way."""
    asyncio.run(main())

diagram - 1 - filter compact

This mode is always agood starting point shows meaningful defaults

diagram - 2 - no filter

This diagram cannot bee shown here (due to image restrictions), but holds all resources from the app apart from IAM Policies.

diagram - 3 - focus on API Gateway

This daigram uses a RegEX filter to strip down the diagramm to all resource starting with r"AWS::ApiGateway::*"

Due to the current architecture of the plugin, the entire process of CDK synthesis needs to be asynchronous. Based on the results of the process, the pictures are generated. From the code perspective, you need to implement a main() function around app = cdk.App() and app.synth(). With that, we add graph = CdkGraph( … ) and after the app.synth() command, the graph.report() command. Finally, we introduce a boilerplate to call the main() safely and asynchronously.

As you can see in this example, we can leverage a wide range of filters as shown here: CDK Graph Plugin Diagram. These filters allow you to create one or multiple versions of visual representations of your architecture. The filters can be explicit, static, or design-based regular expressions.

Final Words

The plugin can be implemented quickly, increases the understanding of the CDK app, and brings additional value regarding the documentation of your work for third-party people. I found two situations where it’s currently not working. Some L3 Constructs, like those from AWS Service Catalog, do not support image creation, as the graph creation process breaks due to multiple steps in the sequence of the app creation. You can find the issue here: GitHub Issue. Besides that issue, you may face errors while including predefined CloudFormation stacks in your CDK app.

All in all, the package adds additional value to our pull request reporting, and if you are using CDK, take a look. And if you want to go even futher check out cdk-nag and the Cdk Graph Threat Composer Plugin. Security guys will love you.

Happy Coding :-)

GenAI in the Field of Data Cleansing: First Steps

Dr. Malte Polley — Mon, 28 Oct 2024 07:21:06 +0000

TLDR;

In this post, we dive into how Generative AI (GenAI) can supercharge data cleansing, specifically for cleaning up email addresses and birthdays. We tackle the usual headaches of using regular expressions and show how GenAI can step in to make things easier and faster. You’ll find code snippets for generating fake data, adding some errors, and using Amazon Bedrock for smart error-fixing. The results? GenAI nailed all the birthday fixes, but some emails tripped up due to special characters.

Introduction: PowerUp for RegEX Implementation

GenAI has been discussed in many different fields. For a long time, I have been thinking about how to implement GenAI in Data Preparation or Integration. So far, I have tried to implement regular expressions to first identify invalid values in a table. In a second step, I attempted to build a rule set using Spark, Python, or SQL to correct inaccurate values.

Both steps are tedious due to edge cases and different development streams. In any case, the work was valuable, as we see significant improvements in data analysis and the creation of data products. Ultimately, the harmonization of data sources will always lead to opportunities in every organization.

Due to this effort and the value of data cleansing, I want to share my first steps in leveraging GenAI to reduce the implementation effort. This is possible because a prompt describes a generic task, like "fix this string according to ... if it contains errors, mistakes, and typos," and the GenAI model in the background will decide on its own how to fix the string. By using such an approach, we combine regex development and error correction in one step. One caveat might be that the response behavior may vary from time to time.

Creating a Sample: Cleaning Email Addresses and Birthdays

To showcase my experiences so far, we will create samples coming from an internal system. It will contain data such as emails and birthdays from our customers. Unfortunately, our input system does not check the data that customers enter into our system. Therefore, customers occasionally send incorrect data to our CRM.

Data Producer: Creating a Ground Truth for Evaluation Purposes

Our system will be mocked by this Python script. This will create valid or expected inputs with 100% accuracy. Afterwards, we will introduce typical errors and typos with a 50% chance into the entries of our dataset. Let's start with emails.

# src/dataproducer.py

def generate_random_email(self):
    first_name = random.choice(self.first_names)
    last_name = random.choice(self.last_names)
    domains = ['example.com', 'test.com', 'sample.com', 'demo.com']
    email = f"{first_name}.{last_name}@{random.choice(domains)}"
    return email.lower()

def introduce_typo_email(self, email: str):
    if random.random() < 0.5:
        email_list = list(email)
        index = random.randint(0, len(email) - 1)
        action = random.choice(['replace', 'add', 'remove', 'case_change'])
        if action == 'replace':
            email_list[index] = random.choice('abcdefghijklmnopqrstuvwxyz')
        elif action == 'add':
            email_list.insert(index, random.choice('abcdefghijklmnopqrstuvwxyz'))
        elif action == 'remove' and len(email) > 1:
            email_list.pop(index)
        elif action == 'case_change':
            email_list[index] = email_list[index].upper() if email_list[index].islower() else email_list[index].lower()
        return ''.join(email_list)
    return email

The function generate_random_email() creates random emails based on a predefined set of German first and last names and four different email providers. The function introduce_typo_email() introduces an error with a 50% chance by replacing, adding, or removing a letter from an email. Letters can also be changed to uppercase.

The second data field will be the birthday of our customers.

def generate_correct_birthday(self):
        start_date = datetime(1900, 1, 1)
        end_date = datetime(2023, 12, 31)
        random_date = start_date + timedelta(days=random.randint(0, (end_date - start_date).days))
        return random_date.strftime("%d.%m.%Y")

def introduce_typo_birthday(self, birthday):
    if random.random() < 0.5:
        birthday = birthday.lstrip('0')
    if random.random() < 0.5:    
        birthday = birthday.replace('.', random.choice([' ', '/', '-']))
    if random.random() < 0.5:
        if "19" in birthday or "20" in birthday:
            year = birthday.split('.')[-1]
            if year.startswith("19") and random.choice([True, False]):
                year = year[2:]
            elif year.startswith("20") and random.choice([True, False]):
                year = year[2:]
            birthday = '.'.join(birthday.split('.')[:-1] + [year])
    return birthday

The function generate_correct_birthday() creates birthdays in the format DD.MM.YYYY. Each date will be between 01.01.1900 and 31.12.2023. The typos we introduce can include changing the separator from "." to "/" or "-" or even removing any separator. We will remove leading zeros or "19" and "20".

Finally, we will write down the dataset via CSV. For this blog post, we will limit the number of rows to 100.

def create_pandas_dataframe(self):
    emails_correct = [self.generate_random_email() for _ in range(self.rows)]
    emails_typo = [self.introduce_typo_email(email=email) for email in emails_correct]
    birthdays_correct = [self.generate_correct_birthday() for _ in range(self.rows)]
    birthdays_typo = [self.introduce_typo_birthday(birthday=birthday) for birthday in birthdays_correct]
    data = {
        'EmailCorrect': emails_correct,
        'EmailTypo': emails_typo,
        'BirthdayCorrect': birthdays_correct,
        'BirthdayTypo': birthdays_typo,
    }
    return pd.DataFrame(data)

def write_as_csv(self, dataframe: pd.DataFrame):
    dataframe.to_csv(path_or_buf='data/out.csv', index=False)

We execute the snippets above from the file src/datacleaner.py.

# src/datacleaner.py
# snippet

def create_data():
    logging.info("Creating DataFrame")
    data_producer = dp.DataProducer(
        rows=100,
        first_names=['Max', 'Sophie', 'Leon', 'Marie', 'Paul', 'Emma', 'Lukas', 'Hannah', 'Tim', 'Anna'],
        last_names=['Müller', 'Schmidt', 'Schneider', 'Fischer', 'Weber', 'Meyer', 'Wagner', 'Becker', 'Hoffmann', 'Schulz'],
    )
    df = data_producer.create_pandas_dataframe()
    data_producer.write_as_csv(dataframe=df)
    logging.info("Wrote down DataFrame as csv")

if __name__ == "__main__":
    create_data()

GenAI-based Data Cleansing: Invoking Claude Haiku for Dynamic RegEX Evaluation at Runtime

To implement a dynamic GenAI-based RegEX at runtime, we need to handle four things. First, we need to create a prompt that holds the relevant information. Second, we need to prepare the data so it can be processed by our GenAI model. Finally, we need to invoke the Amazon Bedrock endpoint. Lastly, the result from the endpoint needs to fit and be passed into our DataFrame.

Prompting, Retrieving, and Running the Implementation

I followed this website to create the prompt. The following snippet shows how to create dynamic prompts for different data fields. I marked the instructions and the data with XML tags as suggested on the website.

def prompt_builder(field_name: str, pattern: str, data: list[str]):
    prompt = f"""<instructions>You are in charge of ensuring data quality. 
Check the following data fields. They represent {field_name} 
from our CRM of our German customers. Fix all fields if they show typos or other mistakes. Ensure your answers follow this pattern: {pattern}.
Return only a comma-separated list and keep the order of the input data.</instructions> <data>{data}</data>"""

    return prompt

The idea is to create a role-based approach to ensure that the task is clearly and directly formulated, as well as being precise about the expected output and, in our case, also the order.

The comma-separated list is a preparation for integrating the results into our existing DataFrame. To do this, I implemented the following code:

def process_output(bedrock_response: dict, json_parsable: bool = False):
    output = bedrock_response['body'].read()
    output = output.decode('utf-8')
    output = json.loads(output)['content'][0]['text']
    if json_parsable:
        output = json.loads(output)
    return [elem.strip() for elem in output.split(",")]

It's important to mention that the result is expressed in natural language. That's why I need to process the output a little bit to extract elements from the text. But the processing is relatively straightforward, as it involves removing leading empty spaces in front of each word (" 05.12.1988").

Now that we have declared the input and output handling, we just need to invoke the endpoint:

def invoke_endpoint(bedrock_client: boto3.client, prompt: str, top_k: int, max_tokens: int, top_p: int, temperature: int):
    body = json.dumps(
        {
            'messages': [{'role': 'user', 'content': [{'type': 'text', 'text': prompt}]}],
            'anthropic_version': 'bedrock-2023-05-31',
            'max_tokens': max_tokens,
            'temperature': temperature,
            'top_p': top_p,
            'top_k': top_k,
        }
    )
    modelId = 'anthropic.claude-3-haiku-20240307-v1:0'
    accept = "*/*"
    contentType = "application/json"
    response = bedrock_client.invoke_model(
        body=body,
        modelId=modelId,
        accept=accept,
        contentType=contentType,
    )
    return response

Et voilà, the final script:

# src/datacleaner.py
# snippet

def invoke_endpoint(bedrock_client: boto3.client, prompt: str, top_k: int, max_tokens: int, top_p: int, temperature: int):
    body = json.dumps(
        {
            'messages': [{'role': 'user', 'content': [{'type': 'text', 'text': prompt}]}],
            'anthropic_version': 'bedrock-2023-05-31',
            'max_tokens': max_tokens,
            'temperature': temperature,
            'top_p': top_p,
            'top_k': top_k,
        }
    )
    modelId = 'anthropic.claude-3-haiku-20240307-v1:0'
    accept = "*/*"
    contentType = "application/json"
    response = bedrock_client.invoke_model(
        body=body,
        modelId=modelId,
        accept=accept,
        contentType=contentType,
    )
    return response

def prompt_builder(field_name: str, pattern: str, data: list[str]):
    prompt = f"""<instructions>You are in charge of ensuring data quality. 
Check the following data fields. They represent {field_name} 
from our CRM of our German customers. Fix all fields if they show typos or other mistakes. Ensure your answers follow this pattern: {pattern}.
Return only a comma-separated list and keep the order of the input data.</instructions> <data>{data}</data>"""

    return prompt

def process_output(bedrock_response: dict, json_parsable: bool = False):
    output = bedrock_response['body'].read()
    output = output.decode('utf-8')
    output = json.loads(output)['content'][0]['text']
    if json_parsable:
        output = json.loads(output)
    return [elem.strip() for elem in output.split(",")]

def gen_ai_processing(create_data: bool, aws_region: str, data_path: str):
    config = Config(
        region_name=aws_region,
    )

    logging.info("Creating or reading data ...")
    if create_data:
        create_data()
        logging.info("Created data ...")
    else:
        df = pd.read_csv(filepath_or_buffer=data_path)
        logging.info("Read data ...")

    bedrock_client = boto3.client("bedrock-runtime", config=config)

    logging.info("Running Bedrock inference for data cleansing")
    for entry in [
        {
            "FieldName": "Email",
            "Pattern": "first_name.last_name@company.com",
            "ColumnTypo": "EmailTypo",
            "ColumnCorrect": "EmailCorrect"
        },
        {
            "FieldName": "Birthday",
            "Pattern": "DD.MM.YYYY",
            "ColumnTypo": "BirthdayTypo",
            "ColumnCorrect": "BirthdayCorrect"
        }
    ]:
        prompt = prompt_builder(
            field_name=entry["FieldName"],
            pattern=entry["Pattern"],
            data=df[entry["ColumnTypo"]].tolist()
        )

        logging.info(f"Running Bedrock inference for {entry['FieldName']}")

        llm_res = invoke_endpoint(
            bedrock_client=bedrock_client,
            prompt=prompt,
            temperature=0.2, 
            top_k=100, 
            top_p=0.2, 
            max_tokens=1024
        )

        processed_output = process_output(bedrock_response=llm_res, json_parsable=False)

        df[f"{entry['FieldName']}Bedrock"] = processed_output

        comparison_column = np.where(df[f"{entry['FieldName']}Bedrock"] == df[entry["ColumnCorrect"]], True, False)
        df[f"CompareBedrockOrg{entry['FieldName']}"] = comparison_column

    df.to_csv(path_or_buf='data/bedrock_out.csv', index=False)  

if __name__ == "__main__":
    gen_ai_processing(
        create_data=False,
        aws_region="eu-central-1",
        data_path="data/out.csv",
    )

Analyzing the Results

Now it's time to conduct an analysis on our dynamic RegEX implementation with GenAI. Since we know the Ground Truth, we can simply compare the results from Bedrock with it. Initially, we have 100 entries for birthdays and emails. The script created 45 emails and 31 birthdays with typos.

With Bedrock, all birthdays could be fixed:

INFO:root:   BirthdayCorrect BirthdayTypo BirthdayBedrock  CompareBedrockOrgBirthday
0       06.06.2006    6/06/2006      06.06.2006                       True
5       22.12.1957   22 12 1957      22.12.1957                       True
6       23.03.1985     23.03.85      23.03.1985                       True
11      22.07.2010   22.07.2010      22.07.2010                       True
14      28.03.1970   28/03/1970      28.03.1970                       True

While 5 emails could not be fixed:

INFO:root:                EmailCorrect                 EmailTypo              EmailBedrock  CompareBedrockOrgEmail
0       tim.weber@sample.com       ti.weber@sample.com       ti.weber@sample.com                   False
5      marie.müller@test.com     marie.müller@test.com    marie.mueller@test.com                   False
6   sophie.hoffmann@demo.com   sophie.offmann@demo.com   sophie.offmann@demo.com                   False
11    hannah.müller@test.com   hannah.müller@testf.com   hannah.mueller@test.com                   False
14   paul.müller@example.com  payul.müller@example.com  paul.mueller@example.com                   False

However, if you look closer, you can see that Bedrock fixed 3 out of 5 correctly, as emails cannot contain German special characters. Thus, two names ("Tim" and "Hoffmann") could be corrected.

Final Words

I was very impressed by this approach. However, I am fully aware of some limitations of this blog post. The data set is small. In larger settings, chunking will be crucial as Bedrock has strict limits for batch approaches. Also, the examples are relatively simple. More complex data fields might show lower success rates. The reproduction showed very good results in five rounds; I consistently got the same results here. However, in larger sets, this needs to be evaluated.

I hope you gained some insight into GenAI-based RegEX implementation for data cleansing. Feel free to reach out if you want to discuss! I will be happy! :-)

Happy coding!

Exploring the Matillion API in Data Productivity Cloud (DPC): Generic Pipeline Monitoring as well as Execution

Dr. Malte Polley — Mon, 16 Sep 2024 09:25:32 +0000

Handling pipeline monitoring can often feel like a daunting task, especially when it comes to managing pipeline dependencies. For small teams, gaining visibility into pipeline executions is essential, yet challenging. Having worked with Matillion for over a year, I’ve encountered a persistent issue: obtaining comprehensive logs with minimal effort.

While you can utilize SNS components for logging each step, this approach demands numerous components for precise logging. Alternatively, custom scripts can be created to connect with third-party solutions. However, the absence of a built-in generic logging notification option in Matillion DPC has always been a drawback.

In this article, I will demonstrate how to leverage the DPC API to initiate and track pipelines, paving the way for event-based child pipeline executions.

APIs are there to rule the IT World – also within Matillion

APIs have become a cornerstone of modern IT solutions, and fortunately, Matillion offers a robust API (LINK) that allows you to initiate pipeline executions seamlessly. As we delve deeper into the API documentation, we discover the necessity of creating a technical user. This user will receive a CLIENT_ID and a CLIENT_SECRET, which are pivotal before you can start working with the DPC. To kick off our journey, we need to obtain a BEARER token:

curl --location 'https://id.core.matillion.com/oauth/dpc/token' \
--header 'Content-Type: application/x-www-form-urlencoded' \
--data-urlencode 'grant_type=client_credentials' \
--data-urlencode 'client_id=<CLIENT_ID>' \
--data-urlencode 'client_secret=<CLIENT_SECRET>' \
--data-urlencode 'audience=https://api.matillion.com'

The response from the DPC API provides the much-needed BEARER token:

{
  "access_token": "eyJhbGciOiJSUzI1NiIsInR5cCIgOiAiSldUIiwia2lkIiA6ICJYTzUtTWtvM0hPYWtJRkdIeXNCSFp2RnQ5SElYRzcxWmhudlJjVnc4UEtvIn0...",
  "expires_in": 1800,
  "refresh_expires_in": 0,
  "token_type": "Bearer",
  "not-before-policy": 0,
  "scope": "pipeline-execution"
}

Matillion DPC provides extensive API documentation, including a comprehensive tutorial for executing and tracking pipeline executions: Executing and managing a pipeline - Matillion Docs.

Translation into Python Scripts

Armed with the tutorial, we can create a Python class that streamlines our interaction with the Matillion API. This class will encompass several key tasks:

Authentication and Interaction: Manage authentication and API interactions.
Initialization: Accept client ID and secret as arguments to set up URLs for token retrieval and API access.
Token Request: Implement a method to obtain a Bearer token for API access.
Project Data Retrieval: Create a method to fetch project data from Matillion, crucial for tracking pipeline status later on. Pipeline Execution: Trigger the execution of a specified pipeline, updating the project data with the execution ID and status.
Pipeline Status Check: Monitor the status of the pipeline execution and update the project data accordingly.

As we examine the code, we notice that the script iterates through responses extensively. This can be streamlined in your projects. I utilize these scripts in conjunction with AWS CDK and AWS Lambda, which necessitates multiple iterations to pinpoint the correct pipeline.

Ultimately, we aim to receive notifications about errors for every pipeline execution. Thus, we’ve integrated Microsoft Teams to facilitate error logging and provide a direct link to the relevant Matillion Dashboard:

Notification Method: Sends a message regarding the pipeline execution status to a designated Teams channel via a webhook.

class MatillionCheck(requests.auth.AuthBase):
    """Matillion DPC API class.

    Args:
        requests (Object): requests.auth.AuthBase class
    """

    def __init__(self, client_id: str, client_secret: str):
        """Initiate Matillion DPC API class.

        Args:
            client_id (str): Matillion client id
            client_secret (str): Matillion client secret
        """
        self.client_id = client_id
        self.client_secret = client_secret
        self.acces_token_url = "https://id.core.matillion.com/oauth/dpc/token"
        self.base_url = "https://eu1.api.matillion.com/dpc"

    def request_api_access_token(self, verify_ssl: bool):
        """Request a Bearer token from Matillion API.

        Args:
            verify_ssl (bool): secure transport flag

        Returns:
            access_token: Bearer token
        """
        headers = {"Content-Type": "application/x-www-form-urlencoded"}
        data = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "audience": "https://api.matillion.com",
        }

        response = requests.post(
            self.acces_token_url, headers=headers, data=data, verify=verify_ssl
        )
        return response.json()["access_token"]

    def get_project_data(
        self,
        access_token: str,
        verify_ssl: bool,
        target_project_name: str,
        target_environment_name: str,
        target_pipeline_name: str,
    ):
        """Get project data form Matillion API.

        Args:
            access_token (str): Bearer token
            verify_ssl (bool): secure transport flag
            target_environment_name (str): name of the environment
            target_pipeline_name (str): name of the pipeline name

        Returns:
            project_data: data collection
        """
        url_projects_ids = self.base_url + "/v1/projects"
        url_environments = self.base_url + "/v1/projects/{projectId}/environments"
        url_published_pipelines = (
            self.base_url + "/v1/projects/{projectId}/published-pipelines"
        )
        headers = {
            "Authorization": f"Bearer {access_token}",
            "Accept": "application/json",
        }
        params = {"size": 100, "page": 0}
        projects_left = True

        while projects_left:
            try:
                projects = requests.get(
                    url_projects_ids, headers=headers, params=params, verify=verify_ssl
                )
                logging.info(f"Check for API errors: {projects.raise_for_status()}")
                projects_json = projects.json()
                if len(projects_json["results"]) > 0:
                    logging.info(f"Found {len(projects_json['results'])} projects")
                    for i in projects_json["results"]:
                        if target_project_name == i["name"]:
                            logging.info(
                                f"Gathering information on project name {i['name']} with id {i['id']}"
                            )

                            environments = requests.get(
                                url_environments.replace("{projectId}", i["id"]),
                                headers=headers,
                                verify=verify_ssl,
                            )
                            logging.info(
                                f"Check for API errors: {environments.raise_for_status()}"
                            )
                            environments_json = environments.json()
                            for n in environments_json["results"]:
                                if target_environment_name in n["name"]:
                                    environment_name = n["name"]
                                    logging.info(
                                        f"Found target environment {n['name']}"
                                    )
                            params["environmentName"] = environment_name
                            published_pipelines = requests.get(
                                url_published_pipelines.replace("{projectId}", i["id"]),
                                params=params,
                                headers=headers,
                                verify=verify_ssl,
                            )
                            logging.info(
                                f"Check for API errors: {published_pipelines.raise_for_status()}"
                            )
                            published_pipelines_json = published_pipelines.json()
                            for p in published_pipelines_json["results"]:
                                if target_pipeline_name in p["name"]:
                                    published_pipeline_name = p["name"]
                                    logging.info(
                                        f"Found target pipeline {published_pipeline_name}"
                                    )

                            project_data = {
                                "project_id": i["id"],
                                "project_name": i["name"],
                                "environment_name": environment_name,
                                "agent_id": n["defaultAgentId"],
                                "agent_name": n["defaultAgentName"],
                                "published_pipeline_name": published_pipeline_name,
                            }
                            return project_data
                    params["page"] = params["page"] + 1
                else:
                    break
            except Exception as err:
                logging.exception(f"Error: {err}")
                raise
        return True

    def execute_pipeline(self, project_data: dict, access_token: str, verify_ssl: bool, log_group_name: str, log_stream: str):
        """Execute a Matillion pipeline via API.

        Args:
            access_token (str): Bearer token
            verify_ssl (bool): secure transport flag
            project_data (dict): data collection

        Returns:
            project_data: extended data collection
        """
        url_execution = self.base_url + "/v1/projects/{projectId}/pipeline-executions"
        headers = {
            "Authorization": f"Bearer {access_token}",
            "Accept": "application/json",
        }

        body = {
            "pipelineName": project_data["published_pipeline_name"],
            "environmentName": project_data["environment_name"],
        }
        try:
            execution = requests.post(
                url_execution.replace("{projectId}", project_data["project_id"]),
                json=body,
                headers=headers,
                verify=verify_ssl,
            )
            logging.info(f"Check for API errors: {execution.raise_for_status()}")
            execution_json = execution.json()
            logging.info(execution_json)
            execution_id = execution_json["pipelineExecutionId"]
            project_data["last_execution_id"] = execution_id
            project_data["execution_status"] = "RUNNING"
        except Exception as err:
            logging.exception(f"Error: {err}")

        return project_data

    def get_pipeline_status(
        self,
        project_data: dict,
        access_token: str,
        verify_ssl: bool,
    ):
        """Get Matillion pipeline execution status.

        Args:
            project_data: data collection
            access_token (str): Bearer token
            verify_ssl (bool): secure transport flag

        Returns:
            project_data: data collection
        """
        url_execution_status = (
            self.base_url
            + "/v1/projects/{projectId}/pipeline-executions/{pipelineExecutionId}"
        )
        url_execution_status_steps = (
            self.base_url
            + "/v1/projects/{projectId}/pipeline-executions/{pipelineExecutionId}/steps"
        )
        headers = {
            "Authorization": f"Bearer {access_token}",
            "Accept": "application/json",
        }
        pipeline_execution_id = project_data["last_execution_id"]
        project_id = project_data["project_id"]
        body = {
            "pipelineName": project_data["published_pipeline_name"],
            "environmentName": project_data["environment_name"],
        }
        try:
            response = requests.get(
                url_execution_status.replace("{projectId}", project_id).replace(
                    "{pipelineExecutionId}", pipeline_execution_id
                ),
                json=body,
                headers=headers,
                verify=verify_ssl,
            )
            logging.info(f"Check for API errors: {response.raise_for_status()}")
            response_json = response.json()
            logging.info("Logging response pipeline status")
            logging.info(response_json)
            if response_json["result"]["status"] == "RUNNING":
                project_data["execution_status"] = response_json["result"]["status"]
            elif response_json["result"]["status"] == "FAILED":
                project_data["execution_status"] = response_json["result"]["status"]
                try:
                    project_data["execution_message"] = response_json["result"][
                        "message"
                    ]
                except KeyError as err:
                    logging.exception(err)
                    response = requests.get(
                        url_execution_status_steps.replace(
                            "{projectId}", project_id
                        ).replace("{pipelineExecutionId}", pipeline_execution_id),
                        params={"size": 100, "page": 0},
                        json=body,
                        headers=headers,
                        verify=verify_ssl,
                    )
                    logging.info(f"Check for API errors: {response.raise_for_status()}")
                    response_json = response.json()
                    logging.info("Logging response pipeline status on steps")
                    logging.info(response_json)
                    for i in response_json["results"]:
                        if i["result"]["status"] == "FAILED":
                            try:
                                project_data[
                                    "execution_message"
                                ] = f"""Step {i["name"]} failed with error {i["result"]["message"]}"""
                            except KeyError:
                                break
                        else:
                            project_data[
                                "execution_message"
                            ] = "Unknown error. Look into Matillion Dashboard."
            else:
                project_data["execution_status"] = response_json["result"]["status"]
        except Exception as err:
            logging.exception(f"Error: {err}")
            raise
        return project_data

    def send_notifcation(self, event: dict, log_group_name: str, log_stream: str):
        """Send a message about the current rotation status.

        Args:
            event (dict): Message dict to be sent via Teams
            log_group_name (str): Name of the CWL Log Group
            log_stream (str): Name of the Log Stream
        """
        try:
            project_name = event["project_name"]
            published_pipeline_name = event["published_pipeline_name"]
            last_execution_id = event["last_execution_id"]
            execution_message = event["execution_message"]
        except KeyError as e:
            logging.exception(e)
            raise

        logging.info("Starting webhook message to Teams")
        try:
            msg = {
                "type": "adaptiveCard",
                "attachments": [
                    {
                        "contentType": "application/vnd.microsoft.card.adaptive",
                        "content": {
                            "$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
                            "type": "AdaptiveCard",
                            "version": "1.5",
                            "body": [
                                {
                                    "type": "Image",
                                    "url": "https://docs.matillion.com/assets/mtln-docs-logo-white.png",
                                    "altText": "Matillion",
                                },
                                {
                                    "type": "Container",
                                    "style": "attention",
                                    "bleed": True,
                                    "spacing": "None",
                                    "items": [
                                        {
                                            "type": "TextBlock",
                                            "text": f"**Abbruch im Projekt: {project_name}**",
                                        }
                                    ],
                                },
                                {
                                    "type": "Container",
                                    "style": "emphasis",
                                    "items": [
                                        {
                                            "type": "TextBlock",
                                            "text": f"**Umgebung**: {published_pipeline_name}",
                                        },
                                        {
                                            "type": "TextBlock",
                                            "text": f"**Fehler**: {execution_message}",
                                        },
                                    ],
                                },
                            ],
                            "actions": [
                                {
                                    "type": "Action.OpenUrl",
                                    "title": "Matillion Dashboard zur Pipeline öffnen",
                                    "url": f"https://mrht.observability-dashboard.eu1.matillion.com/pipeline/{last_execution_id}",
                                }
                            ],
                        },
                    }
                ],
            }
            execution = requests.post(
                web_hook,
                json=msg,
                verify=True,
            )
            logging.info(f"Check for API errors: {execution.raise_for_status()}")
        except Exception as err:
            logging.exception(f"Error: {err}")
            raise
        logging.info("Sent message to Teams")

In the event of an error within Matillion, you’ll receive a message in Teams that looks something like this:

The button “Matillion Dashboard zur Pipeline öffnen” will direct you straight to the corresponding execution dashboard within DPC.

Using AWS to Execute Our Matillion Monitoring Handler

With our code in place, we can leverage an AWS Lambda function in conjunction with AWS Step Functions to execute and monitor the status of a Matillion pipeline execution. AWS Lambda serves as a code execution service, while AWS Step Functions act as a serverless workflow manager and state machine.
The workflow for our DPC Pipeline Execution and monitoring service is depicted below:

This workflow, managed by AWS Step Functions, begins with invoking a Lambda function named "SubmitJob-matrix-42-it-reporting." Following this, a wait state called "Wait30Seconds-matrix-42-it-reporting" pauses execution for 30 seconds. Subsequently, another Lambda function, "GetStatus-matrix-42-it-reporting," checks the job's status. A choice state, "JobComplete-matrix-42-it-reporting," evaluates the execution status. If the status is "FAILED," the workflow transitions to a fail state; if "SUCCESS," it moves to a succeed state, marking the completion of the workflow.

We can utilize the same Lambda function to execute the states “SubmitJob” and “GetStatus.” By using environment variables, we can create generic code to identify the correct Matillion resource for this workflow. Here’s a snippet of the generic Lambda:

def main(event, context, secretsmanager=secretsmanager_client):
    """Orchestrate all other functions.

    Args:
        event (object): Lambda event object
        context (object): Lambda context object
        secretsmanager (boto3, optional): secretsmanager boto3 client object. Defaults to secretsmanager_client.

    Returns:
        project_data (dict): data collection
    """
    logging.info("Starting Matillion Execution Management")
    logging.info(event)
    logging.info(context)
    try:
        secret_arn = os.environ["CLIENT_SECRET_ARN"]
        target_pipeline_name = os.environ["TARGET_PIPELINE_NAME"]
        target_environment_name = os.environ["TARGET_ENVIRONMENT_NAME"]
        target_project_name = os.environ["TARGET_PROJECT_NAME"]
    except KeyError as e:
        logging.exception(e)
        lambda_notifcation(message=e, project_name=target_project_name, log_group_name=log_group_name, log_stream=log_stream)
        raise e

    try:
        response = secretsmanager.get_secret_value(SecretId=secret_arn)
        secret_dict = json.loads(response["SecretString"])
    except Exception as err:
        logging.exception(err)
        lambda_notifcation(message=err, project_name=target_project_name)
        raise err

    matillion = MatillionCheck(
        client_id=secret_dict["client_id"], client_secret=secret_dict["client_secret"]
    )

    logging.info("Getting Bearer Token")
    access_token = matillion.request_api_access_token(verify_ssl=True)
    logging.info("Got Bearer Token")

    try:
        execution_status = event["execution_status"]
        last_execution_id = event["last_execution_id"]
        published_pipeline_name = event["published_pipeline_name"]
        environment_name = event["environment_name"]
        project_data = event
    except KeyError:
        logging.info("Starting new pipeline execution")
        logging.info("Searching for project data ...")
        project_data = matillion.get_project_data(
            access_token=access_token,
            verify_ssl=True,
            target_project_name=target_project_name,
            target_environment_name=target_environment_name,
            target_pipeline_name=target_pipeline_name,
        )
        if project_data is not True:
            logging.info(f"Found project: {project_data} ...")

            logging.info("Executing pipelines ...")
            project_data = matillion.execute_pipeline(
                project_data=project_data, 
                access_token=access_token, 
                verify_ssl=True,
            )
            logging.info(f"Executed pipeline: {project_data} ...")

            published_pipeline_name = project_data["published_pipeline_name"]
            environment_name = project_data["environment_name"]
            last_execution_id = project_data["last_execution_id"]
            execution_status = project_data["execution_status"]

    logging.info(
        f"Checking pipeline status in project {target_project_name} for pipeline {published_pipeline_name} in environment {environment_name}"
    )
    logging.info(f"Last execution: id {last_execution_id} - status {execution_status}")
    project_data = matillion.get_pipeline_status(
        project_data=project_data,
        access_token=access_token,
        verify_ssl=True,
    )

    if project_data["execution_status"] == "FAILED":
        logging.exception(project_data["execution_message"])
        matillion.send_notifcation(
            event=project_data,
        )
    logging.info("Checked Pipeline status ...")

    logging.info("Finished Matillion Execution Management")

    return project_data

This function orchestrates the lifecycle of a Matillion pipeline execution, managing both new executions and status checks for existing ones.

Handling Matillion Child Pipelines in an Event-Based Way

By implementing Lambda and Step Functions, you can listen within AWS for a successful event of the Step Functions execution. To achieve this, connect Step Functions via Amazon EventBridge Rules. The rule requires a pattern to filter relevant API calls:

{
  "detail-type": ["Step Functions Execution Status Change"],
  "detail": {
    "stateMachineArn": ["arn:aws:states:eu-central-1:ACCOUNT_ID:stateMachine:StateMachinematrix42etl1788E6C1-afV30G0wnBsD"],
    "status": ["SUCCEEDED"]
  },
  "source": ["aws.states"]
}

In this setup, if the status of a specific Step Function is “SUCCEEDED,” a target Step Function will be initiated, allowing for seamless event-driven execution.

Summary

In conclusion, the generic pipeline monitoring feature is essential for Matillion DPC. It simplifies processes for all customers and, on the flip side, I am thrilled about the API's capabilities, which empower us to create remarkable solutions and event-based pipeline dependencies—an invaluable asset.

While I haven’t delved into Infrastructure as Code (IaC) in this blog post, it's certainly a great idea to incorporate it. With the Cloud Development Kit, you can leverage multiple Lambdas and Step Functions to automate your deployment process.

On the Microsoft Teams side, all you need is access to Teams workflows to allow incoming webhook workflows, whether for a channel or group chat. And there you have it—a robust Microsoft Teams and AWS Matillion monitoring integration!

Happy Coding!

Exploring CDK and Policy as Code with CDK-Nag and Python

Dr. Malte Polley — Wed, 11 Sep 2024 12:20:52 +0000

Infrastructure as Code (IaC) has become a standard in cloud development, allowing for quick environment setups and compliance through versioning. Tools like Terraform and the Cloud Development Kit (CDK) simplify the process compared to traditional CloudFormation.

In a previous blog post, I discussed using a custom solution for pull request reporting with cfn-lint and cfn_nag. While cfn-lint is still valuable for creating your own compliance rules, keeping up with CDK and AWS recommendations can be challenging. Fortunately, cdk-nag can serve as a substitute for cfn-lint.

CDK-Nag: Making Infrastructure Decisions Visible

Most tools I've used have a common issue: they can check your infrastructure but often fall short on providing actionable insights. cdk-nag enhances visibility into the compliant aspects of your IaC deployment. It comes with prebuilt rule packs for various standards:

AWS Solutions
HIPAA Security
NIST 800-53 Rev 4
NIST 800-53 Rev 5
PCI DSS 3.2.1

These rules are mapped to specific controls, making compliance evaluations straightforward, even for non-AWS exployees.

Sample Implementation of CDK-Nag in Python

For demonstration, we'll use a sample CDK application created with the command cdk init app --language python. If you're new to CDK, check out the official tutorial.

Here's a minimal example of creating an S3 bucket in your CDK stack:



from aws_cdk import Stack, aws_s3 as s3
from constructs import Construct
from src.load_env.config import CDKConfig

class CdkSampleRepoStack(Stack):
    """Create the actual deployment in each AWS account."""

    def __init__(self, scope: Construct, construct_id: str, stage: str, config: CDKConfig, **kwargs) -> None:
        """Initialize CDK stack class."""
        super().__init__(scope, construct_id, **kwargs)

        # Create the S3 bucket
        bucket = s3.Bucket(self, id="MyBucket")

After running cdk synth, you'll generate a CloudFormation template in the cdk.out folder. However, this bucket configuration may have implicit or missing settings, such as encryption or TLS policies.

To leverage AWS best practices, add cdk-nag to your app.py:



#!/usr/bin/env python3
import os
import cdk_nag
import aws_cdk as cdk
from hello_cdk.hello_cdk_stack import CdkSampleRepoStack

app = cdk.App()
CdkSampleRepoStack(app, "HelloCdkStack", env=cdk.Environment(account='123456789012', region='us-east-1'))
cdk_nag.Aspects.of(app).add(cdk_nag.AwsSolutionsChecks(verbose=True))
app.synth()

Running cdk synth now will display error messages in your terminal, such as:



[Error at /CdkSampleRepoStack/MyBucket/Resource] AwsSolutions-S1: The S3 Bucket has server access logs disabled.

[Error at /CdkSampleRepoStack/MyBucket/Resource] AwsSolutions-S10: The S3 Bucket or bucket policy does not require requests to use SSL.

Handling Best Practice Violations

Sometimes, you may need to explicitly violate a best practice, like disabling access logging for a non-critical bucket. You can suppress specific rules as follows:



# hello_cdk.hello_cdk_stack.py
bucket = s3.Bucket(
    self,
    id="MyBucket",
    block_public_access=s3.BlockPublicAccess(
        block_public_acls=True,
        block_public_policy=True,
        ignore_public_acls=True,
        restrict_public_buckets=True,
    ),
    versioned=True,
    enforce_ssl=True,
    encryption=s3.BucketEncryption.KMS_MANAGED,
)

nag_suppression.add_resource_suppressions(
    construct=bucket,
    suppressions=[
        {
            "id": "AwsSolutions-S1",
            "reason": "This bucket does not hold customer data",
        }
    ],
)

With this suppression, the CSV report will now indicate that the rule has been suppressed. But, cdk-nag not supports suppressions on constructs but also on resource paths and many more

Integrating Pull Request Reporting with cdk-nag

Integrating cdk-nag into your CI/CD pipeline can enhance your deployment process. For instance, in Azure DevOps, you can automate pull request comments based on the cdk synth results.

Here's a brief overview of how to set this up:

Create a message class to handle comments.
Use pandas to read the generated CSV files.
Translate the CSV table to Markdown.
Post comments to your pull requests.



import pull_request_comment
import logging
import pandas as pd
import os

logging.basicConfig(
    level=logging.INFO, format="[%(levelname)s] - %(message)s", force=True
)

def main():
    """Create a Pull Request comment within azure-pipelines-pr.yml"."""
    for filename in os.listdir("./synth/templates/"):
        if filename.endswith(".csv"):
            try:
                df = pd.read_csv(f"./cdk.out/templates/{filename}")
            except FileNotFoundError as e:
                logging.exception(e)
                raise
            if df.empty is False:
                logging.info("Adding CDK Validation report")
                result = msg.add(comment=df.to_markdown())

if __name__ == "__main__":
    main()

In your Azure DevOps pipeline, add a task to run cdk synth and then execute the script to post comments:



- task: AWSShellScript@1
  inputs:
    awsCredentials: ${{ parameters.ServiceConnection }}
    regionName: ${{ parameters.AwsRegion }}
    scriptType: 'inline'
    inlineScript: |
      cdk synth
      python3 src/pull_requests/comment.py

Final Thoughts

Implementing cdk-nag in your workflow may require some effort, but it provides an easier way to ensure compliance before deploying to AWS. Shifting left in your development process is always a good idea, especially for beginners looking to ensure correct deployments.

Happy and save coding! :-)

Integration of Apache Iceberg in S3, Glue, Athena, Matillion, and Snowflake – Part 2

Dr. Malte Polley — Wed, 07 Aug 2024 06:34:30 +0000

TL;DR

In this article, I'll show how I integrated Apache Iceberg with AWS Glue, S3, Athena, Matillion, and Snowflake. We start by collecting data via an API, storing it in S3 as a Parquet file, and then use Glue Crawler and Athena to update the data in an Iceberg table. Matillion orchestrates the entire process, while Snowflake queries the data via an "External Volume." An important step was removing non-ASCII characters from the column names to avoid compatibility issues. In the end, the data can be easily queried in Snowflake. The infrastructure part can be found here.

Working Backwards: The Matillion Orchestration Pipeline

In this case, let's start from the end to see what I implemented regarding Apache Iceberg, Matillion, Snowflake, and AWS. This is the final orchestration pipeline:

As you can see, there isn't much low-code here – and that's actually one of the big benefits of Matillion. Instead, we need to use the Python and Plain SQL components to drive our process.

What's Happening Here?

From Part One, you already know that S3, Glue, and Athena are our AWS powerhouses. The first Python component ("Gather Data") collects data from an API and stores it in S3 as a Parquet file.

# snippet
df = pd.DataFrame.from_dict(response_dict["data"])
df["Upload Date"] = file_date
df.columns = [replace_no_ascii(unicode_string=col) for col in df.columns]
df.to_parquet(
    f"s3://{s3_bucket}/{directory}/{year}/{file_date}.snappy.parquet",
    engine="auto",
    compression="snappy",
    storage_options={
        "key": access_key,
        "secret": secret_access_key,
        "token": sessions_token,
    },
)

The second Python component starts the Glue crawler run and the Athena INSERT process. The Glue step is crucial for maintaining our Glue database, while Athena maintains the actual Glue Database with our Iceberg Table. Here are some code snippets to give you an idea:

def update_glue_database(self, credentials: dict):
    access_key = credentials["Credentials"]["AccessKeyId"]
    secret_access_key = credentials["Credentials"]["SecretAccessKey"]
    session_token = credentials["Credentials"]["SessionToken"]
    glue_client = boto3.client(
        "glue",
        aws_access_key_id=access_key,
        aws_secret_access_key=secret_access_key,
        aws_session_token=session_token,
    )
    try:
        glue_client.start_crawler(
            Name=self.crawler_name
        )
        logging.info("Started Glue crawler run")
        finished = False
        while not finished:
            response = glue_client.get_crawler(
                Name=self.crawler_name
            )
            status = response["Crawler"]["State"]
            last_run = response["Crawler"]["LastCrawl"]
            if status == "READY":
                finished = True
                logging.info("Finished Glue Crawler run")
            if status == "READY" and last_run in ["FAILED", "CANCELLED"]:
                raise ValueError("Glue Catalog failed")
            time.sleep(30)
    except ClientError as e:
        logging.exception(e)
        raise

It's important to create a condition to wait for the Glue Crawler process. Otherwise, our Athena INSERT process wouldn't maintain the Iceberg table properly. Here’s the code snippet for the Athena process:

def maintain_iceberg_with_athena(self, credentials: dict, database: str, s3_input_prefix: str) -> None:
    base_table_name = s3_input_prefix.replace("-", "_")
    access_key = credentials["Credentials"]["AccessKeyId"]
    secret_access_key = credentials["Credentials"]["SecretAccessKey"]
    session_token = credentials["Credentials"]["SessionToken"]
    query = f"""INSERT INTO "{base_table_name}_iceberg" SELECT * FROM "{base_table_name}" WHERE "upload date" = (SELECT MAX("Upload Date") FROM "{base_table_name}");"""
    logging.info("Running update on Iceberg table")
    logging.info(query)
    athena_client = boto3.client(
        "athena",
        aws_access_key_id=access_key,
        aws_secret_access_key=secret_access_key,
        aws_session_token=session_token,
    )
    try:
        response = athena_client.start_query_execution(
            QueryString=query,
            QueryExecutionContext={'Database': database, 'Catalog': 'AwsDataCatalog'},
            ResultConfiguration={'OutputLocation': 's3://aws-athena-query-results-673540176684-eu-central-1/'},
            WorkGroup='primary',
        )
        query_execution_id = response["QueryExecutionId"]
        finished = False
        while not finished:
            response = athena_client.get_query_execution(QueryExecutionId=query_execution_id)
            status = response["QueryExecution"]["Status"]["State"]
            if status == "SUCCEEDED":
                finished = True
                logging.info("Finished Iceberg table update")
            if status in ["FAILED", "CANCELLED"]:
                raise ValueError(f"Updating Iceberg table with query {query} failed")
            time.sleep(3)
    except (ClientError, KeyError) as e:
        logging.exception(e)
        raise

As mentioned in the first part, we need to crawl the data in our S3 landing stage via Glue. The Athena process then creates a new data stage. That’s why I decided to clean up my landing stage after the Athena INSERT statement succeeded. With Matillion, we can use environment variables to store the names of files written to S3. We can then use these variables to delete them again. This process is done by the following script block:

def delete_parquet_files_in_s3(self, credentials: dict, parquet_files: list[str]) -> None:
    access_key = credentials["Credentials"]["AccessKeyId"]
    secret_access_key = credentials["Credentials"]["SecretAccessKey"]
    session_token = credentials["Credentials"]["SessionToken"]

    s3_client = boto3.client(
        "s3",
        aws_access_key_id=access_key,
        aws_secret_access_key=secret_access_key,
        aws_session_token=session_token,
    )

    for file_key in parquet_files:
        try:
            s3_client.delete_object(Bucket=self.s3_bucket_name, Key=file_key)
            logging.info(f"File {file_key} deleted from S3 bucket {self.s3_bucket_name}.")
        except ClientError as e:
            logging.exception(e)
            raise

The final SQL component uses this SQL statement:

ALTER ICEBERG TABLE ICEBERG_TABLE REFRESH;

This command is necessary, although the Snowflake Docs do not actually talk about this manual freshness step.

The "Easy" Part on the Snowflake Side

The infrastructure on the Snowflake side consists of a Warehouse, databases, a user, and its role. The data is integrated via an "External Volume":

An external volume is an account-level Snowflake object that stores an identity and access management (IAM) entity for your external cloud storage. Snowflake uses the external volume to securely connect to your cloud storage to access table data and metadata.

Based on that quote, if you are familiar with "External Stages" and "Storage Integration", this is very comparable. You can find the Snowflake docs here. There is also an excellent tutorial available here.

CREATE OR REPLACE EXTERNAL VOLUME iceberg_external_volume
   STORAGE_LOCATIONS =
      (
         (
            NAME = 'my-s3-us-west-2'
            STORAGE_PROVIDER = 'S3'
            STORAGE_BASE_URL = 's3://<my_bucket>/'
            STORAGE_AWS_ROLE_ARN = '<arn:aws:iam::123456789012:role/myrole>'
            STORAGE_AWS_EXTERNAL_ID = 'iceberg_table_external_id'
         )
      );

If you are using Glue as the catalog, this catalog also needs to be created.

CREATE or REPLACE CATALOG INTEGRATION GLUE_DATABASE_ICEBERG
  CATALOG_SOURCE=GLUE
  CATALOG_NAMESPACE='glue-database-iceberg'
  TABLE_FORMAT=ICEBERG
  GLUE_AWS_ROLE_ARN='arn:aws:iam::012345678910:role/glue-iceberg-role'
  GLUE_CATALOG_ID='012345678910'
  ENABLED=TRUE;

Finally, you need to create our Iceberg table.

CREATE OR REPLACE ICEBERG TABLE ICEBERG_TABLE
  EXTERNAL_VOLUME='S3_EXT_REXX_VOLUME'
  CATALOG='GLUE_DATABASE_ICEBERG'
  CATALOG_TABLE_NAME='GLUE_DATABASE_ICEBERG_TABLE';

Afterwards, the table can be queried, while it is coming from Glue and S3. But maybe you noticed the quotation marks in the header of this section. You might encounter this situation:

Processing aborted due to error 300010:3639435319; incident 2552819.

After contacting support, I realized that the API I queried used non-ASCII characters in their fields, which I converted to column names. As of August 2024, this is neither supported by Snowflake nor does the error logging help identify this issue by itself. That’s why I integrated this function in my S3 upload script:

def replace_no_ascii(unicode_string: str):
    for i in range(0, len(unicode_string)):
        try:
            unicode_string[i].encode("ascii")
        except UnicodeEncodeError:
            unicode_string = unicode_string.replace(unicode_string[i], "%")
    return unicode_string

This function removes non-ASCII characters from column names. By reloading the data with that function, I was able to query my Snowflake tables like a charm.

Closing Words

It took a while to get a clear idea about the process of leveraging data from an API as an Iceberg table. Although the process was tedious, this is what I love about my job. You can now take the fast lane. Cheers! 😊

Integration of Apache Iceberg in S3, Glue, Athena, Matillion, and Snowflake – Part 1

Dr. Malte Polley — Mon, 29 Jul 2024 07:56:26 +0000

TL;DR

If you aim to establish transactional object control in your Data Lake, you should definitely consider Apache Iceberg. However, the path is somewhat rocky and involves manual efforts. AWS Glue, in particular, needs to offer more feature sets. Once you overcome these hurdles, the integration of Iceberg in Snowflake, Glue, and Matillion is as comfortable as dealing with other file formats. This article is less of a technical guide and more of a narrative about my journey from S3 to Glue. This part focuses on infrastructure deployment, while the second part will focus on the Matillion pipeline and Snowflake integration.

Transactional Data Lakes – Why and How?

When working with a Data Lake, the question of how to handle changes and data updates inevitably arises. In Europe, explicit deletion concepts are required. In a database, a data point can be relatively easily identified and updated by deleting or anonymizing it.

Apache Iceberg – Why and How?

There are three frameworks that meet the requirements of a transactional Data Lake: Apache Hudi, Apache Iceberg, and Delta Lake. AWS has published a blog post offering a purely technical comparison: Choosing an open table format for your transactional data lake on AWS.

From my perspective, the relevant aspect was that Apache Iceberg has the most extensive integration with Amazon Athena. Both read and write operations can occur here. In combination with Glue, it is possible to update objects via SQL from AWS without maintaining complex systems.

Automation is Key – But Not Given at the Crucial Point

The following section outlines my project process.

Data Integration: From S3 to Glue to Apache Iceberg

When we started, I quickly realized that we couldn't use CSV or JSON files for Apache Iceberg but rather ORC, Avro, and Parquet. In Matillion, I often use Pandas, so migrating from CSV to Parquet wasn't a big problem. My DataFrame only needed to use a different method and exchange a few parameters.

df = pd.DataFrame.from_dict(response_dict["data"])
df["Upload Date"] = file_date
df.columns = [replace_no_ascii(unicode_string=col) for col in df.columns]
df.to_parquet(
    f"s3://{s3_bucket}/{directory}/{year}/{file_date}.snappy.parquet",
    engine="auto",
    compression="snappy",
    storage_options={
        "key": access_key,
        "secret": secret_access_key,
        "token": sessions_token,
    },
)

The storage options are only needed if you want to write the data cross-account.

In the next step, I thought Glue could easily create Iceberg tables from Parquet files – wrong. So, what can Glue do here? You can build Iceberg tables from existing Glue tables. But here's the problem: you have to provide the table definition, even though the crawler has already created it for the "simple" files. This is a significant lack of automation when I want to build my data pipelines with Infrastructure as Code (IaC).

I then wondered about the Python library PyIceberg. It supports Python and AWS but cannot use S3 as a catalog, only Glue: Creating an Iceberg table on S3 using PyIceberg and Glue catalog.

The attempt to let Glue read Iceberg tables directly also failed. Unfortunately, our ELT tool Matillion cannot directly handle Apache Iceberg as of July 2024.

Data Integration: From S3 to Glue to Apache Iceberg – With the Help of Athena

Technically, I always thought it had to continue with the Glue tables that my crawler had already created. I then discovered that Athena can not only create Apache Iceberg tables via SQL but also generate Iceberg tables using DDL statements without giving a table definition.

Thus, you can indeed create Iceberg tables in AWS via custom resources without table definitions and using "IaC":

CREATE TABLE "glue-database-rexx"."datalake_api_iceberg" 
WITH (table_type = 'ICEBERG', format = 'PARQUET', location = 's3://mybucket/datalake-iceberg/', is_external = false) 
AS SELECT * FROM "glue-database-iceberg"."datalake";

The only downside: you create a copy of the existing Parqute files. So cleanup process and data INSERT mechanism is needed. You will find this in part two of this series.

Closing Words

Infrastructure as Code (IaC) for Apache Iceberg is possible – but with detours. In summary, you can use IaC to create S3 buckets, Glue crawlers, the database, and the Athena query for the initial statement to create the Apache Iceberg catalogs. The path is not very explicit, which is why I wrote it down here. Once I understood this, I could then codify my journey with the Cloud Development Kit.

Guided Data Access Patterns: A Deal Breaker for Data Platforms

Dr. Malte Polley — Wed, 15 May 2024 15:17:48 +0000

I am now sure that the best tech stack with the best people using it is of no use to an organization if it is not clear how access to data and the tech stack is. This blog post should therefore be seen as a supplement to the two previous ones, which were more technical. In the first one, I talked about a Zero-ETL approach for the integration of Snowflake and Amazon S3 and in the second one about the integration of Matillion for complex data transformation and data loading processes.

This blog post is more like a diary entry and addresses the two main problems that I have experienced. However, there are also possible solutions that I will talk about.

No Use Case, No KPIs, No Success

Every use case is about a goal. Yeah right. I would always prioritize the Use case. Simply integrating systems does not help any organization. Even more clearly, a cloud environment just costs money and has no benefit if you don't start from the use case. So, ask yourself, what is your use case or what is your requester's use case.

This doesn't just help with the implementation of a project. It also helps you in the evaluation when it comes to project completion. Therefore: no use case, no KPI, no success assessment.

Even worse: If you only think about the system, you may end up in a data swamp and do not create a logical and meaningful storage system for the synergies that you want to create for later new use cases. And I'm sure you want that. But one after another.

The Issues: Organizational and Technical Ones

If we have a use case, various questions may arise. These can be: Who gives me access to the data? What does the data look like? Do these need to be prepared further so that I can use them for my use case? Is there perhaps already some preparatory work that I can build on? And there are a many more.

Technical questions also arise, especially if you need to deal compliance requirements. How do the data have to be encrypted? How do I get test data with which I can do staging? Which fields contain PII-relevant data? How do these have to be treated technically? We do we integrate networks? And these are just a few examples.

Addressing the Issues

So how can we deal with the questions productively? We design clear access patterns to our data platform. These access patterns address precisely the organizational and technical problems.

What do I mean by access pattern? In principle this means a sequence of well-defined steps. The sequence of steps has different participants. In addition to those involved, the access patterns clearly state who must do what. And finally, we need technical requirements that are operationalized with maximum precision. Finally, the documentation needs to be expanded to complete the project. This is the only way to create synergies. Any by that, documentation is useful for new use cases and for data governance requirements.

Let’s continue with a technical sub-component of a prototypical data platform and suggest for a system that addresses both our governance and synergy requirements. But first let's start with a process definition.

The Predefined Process

Our use case should be implemented on the data platform. Our organization has already defined a process for new data products. Every box in the figure is well defined in terms of who is in charge for what. We are very lucky in this case. This is probably not always the case.

The question shows how we start: a description of the exact requirements must be read by the requester. With this he can consult the existing data catalog. For this to happen successfully, the data catalog must contain not only technical information, but also semantic information. What does that mean? Technical information consists of the source and the data that can be taken from a source. This includes, for example, column names or attributes (such as tags) that are read from AWS. However, not only the AWS databases or S3 buckets should be part of the information set, but also the tables and structures that can be found in the data platform in the data warehouse or at their locations. In summary, the data catalog should be able to technically represent all data artifacts.

That helps, but it's not everything. Because now we still lack information. Which fields are PII relevant? Who is the technical and who is the professional contact for data? What is this data used for? Which existing data products are you used for? Questions of this type of aim at semantic information.

From a governance perspective, the data catalog should show which versions of a data route already existed and how it has changed over time. It is also important to see which fields are specifically being transferred or transformed and how.

In summary: The data catalog enables a holistic view of all data artifacts and their use in an organization. Without these system components, requesters cannot decide whether a source system is already connected or whether this connection needs to be expanded. If our requester is lucky, the required sources are already available, and he can start directly with data product development. This must inevitably end with documentation in the data catalog. If he is unlucky, the source system first must be integrated. And this may mean a long journey that can last for months without proper process descriptions and technical blueprints.

The Technical Description of Sub-Components: Make Your GRCS-Team and Yourself happy

To show an example of what I mean by technical operationalization or blueprint, let's also look at a AWS service: Amazon S3. Widely used and still often not sufficiently configured. However, we managed to agree on the following technical blueprint with our Group Governance, Risk, Security and Compliance group.

Each bucket is accessed exclusively, and here in the example of Glue, from a VPC via a VPC endpoint. The local access control list contains no entries and blocks new ones. The same applies for the global access control list Amazon S3. In addition, every access is recorded via access logging. This process is defined via the configuration of a bucket policy. The same also enforces that you can only speak to the Bucket via TLS 1.2 or higher. The Bucket also has a dedicated KMS Customer Managed Key, which is rotated annually once a year. Finally, the consuming resource requires a role that is only allowed to read and write to this bucket and its key.

This shall be only an example. Please look at all AWS best practices for organization wide configuration. But if we have such a description, we can carry out the deployment with the help of a requirements specification and can use this accordingly in the audit. This also makes it possible to repeatedly keep up with AWS steps by taking new measures and removing old ones if necessary. Regardless of this, individual cases are not discussed because these blueprints have global validity. And of course, this process must be carried out for all sub-components of the data platform.

The Data Catalog: Data Hub Project

If you are new to the data catalog field, you can quickly find AWS Glue on the AWS platform. Unfortunately, AWS Glue does not offer a semantic data catalog and you must allow the user in the AWS GUI, which will be difficult to understand, especially for business users.

There are several commercial providers, but I would definitely recommend Data Hub Project. DataHub Project is an open-source metadata platform that serves as an extensible data catalog and supports data discovery, data observability, and federated governance to address the complexity of the data ecosystem. The data catalog enables the combination of technical, operational and business metadata to provide a 360-degree view of data entities. DataHub makes it possible to pre-enrich important metadata using shift-left practices and respond to changes in real time.

The deployment guide uses Kubernetes by default, but the cluster can also be hosted in AWS with many managed services and in the Elastic Container Service.

The Data Hub Project loading page can be secured via SSO and is responsive. A search helps to find data artifacts and find metadata about them.

Each asset has a detailed view, which is shown here in excerpts. In addition to the table definition, the most common table queries are displayed, for example, as are the individual data fields with tags and example values.

The lineage view then shows how the data artifacts depend on each other and which fields are used. In the example, the product starts in S3 and ends in PowerBI via Snowflake. All this information is imported into the catalog system via ingestion. There are quite a few of them.

The only thing the system can't do (yet) and I'm missing is an alert about updates through Ingestions. For example, far too few new values have arrived through integration, or the values differ significantly from previous values.

Final thoughts

I tried to show what is necessary to create access patterns to a data platform as simply as possible. I am convinced that in addition to a good technical foundation and an up-to-date description of the data platform, its capabilities and possibilities, a good platform team is also necessary to work in a customer-oriented manner. This is the only way to transform the people and your organization to act in a data-driven way.

ELT as Compliance Enabler: Running Steampipe with Matillion Data Productivity Cloud

Dr. Malte Polley — Thu, 25 Apr 2024 09:12:41 +0000

This blog post is the second part of my series about the Modern Data Platform (MDP) of MRH Trowe, a German insurance broker. The first part was about the Zero-ETL integration of Snowflake and S3. This part describes our data loading process using the example of Azure AD reporting with the help of steampipe and Matillion Data Productivity Cloud. Every blog post tells its dedicated story.

TLDR;

With steampipe, data engineers can easily query via SQL. In addition to compliance reporting, many other cases can be mapped. With the Data Productivity Cloud, Matillion, on the other hand, offers a low-code platform that can be deeply integrated into corporate networks via custom agents. This gives the customer various options for control mechanisms regarding data flows and network segmentation.

Compliance Requirement: Know your Azure AD Users

For this blog post we use a simple use case. We need a daily snapshot of our users in Azure AD. The list of users should be made available in a table in Snowflake so that it can be further processed from there. The idea is to continue working with login data to identify inactive users and then delete them from Azure AD.

Use Case: Azure AD Identity Reporting

With this reporting use case in hand, there are various options for implementation. MRH Trowe relies heavily on standardization and automation in all its projects. Accordingly, our Azure AD queries must run automatically every day and, in the best case, be easily adaptable for new use cases.

Azure offers its own SDK for various languages for Azure AD queries. This is also easy to use but requires the data engineer to know a programming language. While looking for another option for querying Software as a Service (SaaS) platforms, we at MRH Trowe came across steampipe.

Intro Steampipe

Steampipe is an open-source project from the company turbot. The claim is very catchy:

select * from cloud; Instantly query your cloud, code, logs & more with SQL. Build on thousands of open-source benchmarks & dashboards for security & insights.

With an SQL interface, steampipe is exactly what a data engineer loves. At the same time, the variety of options available from SaaS platforms is great. Finally, steampipe is relatively fast in the query - in any case fast enough for automation in daily batch runs.

In the background, steampipe was created with Go. The developer guide contains various examples of how you can not only access data, but also manipulate data.

Steampipe offers a plugin for Azure AD and many more. In the first, step we only need the azuread_user table. In addition to this, steampipe offers 14 more, so that many use cases should be able to be represented with this set alone. Ultimately, the documentation also offers a mapping between the plugin's tables and possible compliance guidelines that could be implemented in the organization according to a certain standard.

Matillion Data Productivity Cloud

Before starting the MDP, MRH Trowe didn't have much of a reporting perspective on the existing data. PowerBI was primarily used as a one-size-fits-all tool. PowerBI in combination with an SMB file share is problematic purely from a performance perspective. After Snowflake was chosen as the data warehouse and the data integration between the respective Snowflake DWH and S3 was clarified, the question of data manipulation arose. This is where the Matillion Data Productivity Cloud comes into play, which offers us two key advantages from the automation and integration perspective. On the one hand, it is possible to set up all possible communication paths between Snowflake and the source using ready-made components (SMB, SFTP and APIs, etc.) and Matillion also acts as a scheduler for the data transformation and loading processes. The Matillion jobs themselves are created as low code objects and compile both SQL and can also execute Python and Bash scripts.

The Ease of Access: Meet the (Team) Requirements

In addition to a technological analysis of what needs MDP should be met by MRH Trowe, conceptual requirements also had to be clarified. The two essential ones were the integration into the existing network infrastructure but also the easy, seamless access to data transformations. Matillion meets both needs.

As a SaaS product, Matillion offered the possibility of network integration through custom agents. They periodically poll the Matillion API and execute new jobs, if available. This gives MRH Trowe the opportunity to have precise control over the agents' access and to control them. This scenario is based on hybrid data architecture, where resources can be distributed across all network segments in the cloud or on-premises.

Thanks to the low-code approach, everyone - including IT-savvy department users - can understand how the jobs run or design them themselves. However, Matillion also offers high code users the opportunity to solve complex scenarios via SQL, Python and Bash scripts. In any case, what the flow is doing remains clear.

The Architecture: S3 Buckets, Matillion Custom Agents with Fargate and Snowflake Private Link

The data flow can also be private via the custom agents and in conjunction with the Zero-ETL approach via S3 and Snowflake. What is important here to MRH Trowe is to avoid data transport via the custom agents into the Matillion Cloud.

The diagram shows the AWS architecture. For reasons of clarity, not all options are shown. Matillion can be deployed with high availability in an ECS cluster in private subnets behind NAT gateways (green call direction). This means that the custom agents are protected from external access. The agents' security only requires outbound rules to ensure communication with the Matillion API, the Snowflake Private Link and the source and target systems.

To remain private, we need a gateway or interface endpoints for S3 (red call direction). For Snowflake we need the Business Critical service level in the tenant. The AWS PrivateLink Service can then be created afterwards (blue access direction).

The Dockerfile: Taking Ownership

The integration with the custom agents requires a Docker image, which by default does not have to be touched by the Matillion customer because it does run patch and update cycles itself. However, the own care and maintenance of the Docker image offers the opportunity to carry out configurations yourself. In addition, the release cycle can also be controlled.

Regarding our use case, there is no possibility of installing steampipe from Matillion because the user has no permissions in the Docker container. You could now install an execution resource in addition to the custom agents, or you could configure steampipe yourself in the Docker image.

Below you will find an adaptation of the Matillion image for our use case. It is important that the user in the Docker Container does not have root rights, so paths and folders in particular must be made available to the user agentuser.

ARG matillion_image_tage

FROM public.ecr.aws/matillion/etl-agent:$matillion_image_tage

USER root
# Make important paths available
RUN chown -R agentuser /usr/local/bin
RUN chown -R agentuser /bin/uname
USER agentuser

# Install steampipe as non-root
RUN /bin/sh -c "$(curl -fsSL https://steampipe.io/install/steampipe.sh)"
# Install steampipe plugin(s)
RUN steampipe plugin install azuread

# Extend path with important paths
RUN PATH=$PATH:~/usr/local/bin/steampipe
RUN export PATH=$PATH:~/bin/uname

To be able to communicate with Azure AD, tokens must be available (TenantID, ClientID and ClientSecret). These are made available to steampipe via the pipeline or project's environment variables.

The Pipeline: Keep things simple

In Matillion we use an orchestration pipeline. A bash component creates a steampipe logging and starts a limited query in advance as a DB cache update. In the next step, a Python script component carries out the actual query using subprocess and loads the result into an S3 bucket using pandas. In Snowflake, the bucket is known as the External Stage which will be updated via ALTER STAGE command. The existing table in Snowflake is emptied via TRUNCATE and filled with the current Azure AD user list via COPY INTO command. This ends the pipeline. We use SNS topics for the corresponding error logging to be informed about pipeline interruptions.

The Python script can run steampipe via the subprocess library and with the help of pandas manipulation of the data as well as the cross-account data loads can be carried out.

# Snippet
import pandas as pd
import subprocess

def upload_file(
    query: str,
    directory: str,
    env_vars: dict,
    s3_bucket: str,
    credentials: dict,
):
    """Query data from Azure AD and copy as file to s3.

    Args:
        query (str): steam pipe querey
        directory (str): directory for S3
        env_vars (dict): dictionary with Azure AD tokens
        s3_bucket (str): S3 bucket name
        credentials (dict): AWS credentials

    Returns:
        bool: indicates success
    """
    now = datetime.now()
    year = now.strftime("%Y")
    date = now.strftime("%Y-%m-%d")

    file_name = f"{date}_{directory}.csv"
    access_key = credentials["Credentials"]["AccessKeyId"]
    secret_access_key = credentials["Credentials"]["SecretAccessKey"]
    sessions_token = credentials["Credentials"]["SessionToken"]

    logging.info("Running steampipe query")
    logging.info(f'steampipe query {query} --output=json')
    result = subprocess.run(
        [f"steampipe query \"{query}\" --output=json"],
        env=env_vars,
        shell=True,
        capture_output=True,
        text=True
    )
    logging.info(result.stderr)
    user_dict = json.loads(result.stdout)
    df = pd.DataFrame.from_dict(user_dict) 
    df["Upload Date"] = date

    df.to_csv(
        f"s3://{s3_bucket}/{directory}/{year}/{file_name}",
        quoting=csv.QUOTE_ALL,
        quotechar='"',
        sep=";",
        index=False,
        storage_options={
            "key": access_key,
            "secret": secret_access_key,
            "token": sessions_token,
        },
    )
    logging.info(
        f"Uploaded {file_name} to s3://{s3_bucket}/{directory}/{year}/{file_name}"
    )
    return True

Closing Words

This blog post shows how to use steampipe with Matillion and Snowflake to meet compliance reporting requirements. However, the way there wasn't easy either. The integration of subprocess was not trivial because the Docker Image would have to be modified, the PATH variable for subprocess would have to be known and all environment variables would also have to be passed to the subprocess.run() call.

The first versions of this pipeline showed that subprocess works significantly differently than executing the same command in Bash itself. The problem was solved by having the Bash component of Matillion upfront to execute the same, but limited query.

Due to the database in the background, it is recommended to execute steampipe queries in Matillion - especially complex ones - not in parallel, but sequentially. Unfortunately, it is (still) not possible to host the steampipe database on RDS. Although there is a service mode that enables the integration of external databases. The service mode does not seem to be an appropriate approach as in ECS IPs are changing constantly. But in the end, with Snowflake in the background we can grab flat tables and perform complex transformations later on.

Overall, the interaction between steampipe and Matillion is very valuable for MRH Trowe - and not just for compliance reporting.