Post Syndicated from Pradeep Kumar Nalluri original https://aws.amazon.com/blogs/big-data/pythonoperator-and-bashoperator-now-available-on-amazon-managed-workflows-for-apache-airflow-amazon-mwaa-serverless/

If you run Apache Airflow workflows on Amazon MWAA Serverless, you can now use PythonOperator and BashOperator to run custom code directly in the serverless runtime. Previously, Amazon Managed Workflows for Apache Airflow (Amazon MWAA) Serverless only supported orchestration of AWS services through operators for scheduling tasks, managing dependencies, and handling retries. It did not support running your own Python functions or shell scripts natively. If you needed custom Python logic or shell commands, you had to wrap code in AWS Lambda functions, start Amazon Elastic Container Service (Amazon ECS) tasks, or use other AWS compute services. These alternatives add complexity, cost, and latency to your orchestration pipelines.

With this launch, you can run custom Python functions and shell scripts directly within the serverless task runtime, without requiring additional infrastructure. This means you can now use PythonOperator and BashOperator many data engineering teams rely on for ETL pipelines and data quality checks – without provisioning additional compute.

In this post, we walk through how this feature works and demonstrate a practical example: building a serverless pipeline that converts CSV files to JSON format using a PythonOperator, and verifies the output using a BashOperator. By the end, you will know how to:

• Package a Python module with dependencies and upload it to an Amazon Simple Storage Service (Amazon S3) bucket as a code bundle
• Define a multi-task workflow using the dag-factory compatible YAML
• Create and run a workflow with the AWS Command Line Interface (AWS CLI)
• Verify that your pipeline produced the expected output

How it works

With MWAA Serverless, you can package your custom code, upload it to an Amazon S3 bucket, and reference it when creating a workflow. The service snapshots your code at workflow creation time and uses that snapshot for all subsequent runs of the same workflow version.

Code bundles

A code bundle is the package that contains your custom logic. You package your Python modules or shell scripts and upload them to an Amazon S3 bucket. A code bundle can be:

• A single .py file or .sh bash script (uploaded to an Amazon S3 bucket)
• A ZIP archive containing multiple shell scripts, Python modules and dependencies (up to 250 MB)

Execution model

When you create or update a workflow, MWAA Serverless snapshots your code bundle from an Amazon S3 bucket provided and stores it on the service side. At task execution time, the service uses this snapshot – not the object currently residing in your Amazon S3 bucket – to run your code in an isolated runtime environment.

Python and Bash tasks do not have internet access. They can reach only Amazon S3, Amazon Elastic Container Registry (Amazon ECR), and Amazon CloudWatch, which are the services the runtime requires to operate. To have internet access, configure the workflow with Amazon VPC so that it can go through the provided VPC.

Supported operators

The following table describes the two operators now available in MWAA Serverless.

Security

AWS Key Management Service (AWS KMS) encrypts your code bundles at rest. IAM policies control who can create, update, and trigger the workflows. The execution role scopes what AWS resources your code can access at runtime.

Prerequisites

Before getting started, verify that you have the following resources and tools configured in your AWS account:

• An AWS account with access to Amazon MWAA Serverless
• AWS CLI v2 (latest version) installed and configured. To install or update, see Installing or updating to the latest version of the AWS CLI.
• An Amazon S3 bucket for storing DAG definitions and code bundles
• An IAM role that MWAA Serverless can assume (see the execution role setup below)

Walkthrough: Building a serverless CSV-to-JSON pipeline

In this walkthrough, we build a pipeline that converts CSV files to JSON format – a common data transformation for downstream APIs and analytics systems that consume JSON. The pipeline uses a PythonOperator for the conversion logic and a BashOperator to verify the output. Here is what the pipeline does:

1. Reads a CSV file from an Amazon S3 bucket
2. Converts it to JSON format with column type inference
3. Writes the JSON file back to an Amazon S3 bucket
4. Validates record counts match between source and output

Step 1: Create the execution role

Create an IAM role that your workflow assumes at runtime. The trust policy must allow the airflow-serverless.amazonaws.com service to assume the role:

cat > trust-policy.json << 'EOF'
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "airflow-serverless.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}
EOF

Create the role and attach an inline policy granting least-privilege access to your S3 bucket:

aws iam create-role \
  --role-name MWAAServerlessExecutionRole \
  --assume-role-policy-document file://trust-policy.json

aws iam put-role-policy \
  --role-name MWAAServerlessExecutionRole \
  --policy-name MWAAServerlessAccessPolicy \
  --policy-document '{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "s3:GetObject",
        "s3:PutObject",
        "s3:ListBucket"
      ],
      "Resource": [
        "arn:aws:s3:::amzn-s3-demo-mwaa-data",
        "arn:aws:s3:::amzn-s3-demo-mwaa-data/*"
      ]
    },
    {
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogGroup",
        "logs:CreateLogStream",
        "logs:PutLogEvents",
        "logs:DescribeLogStreams",
        "logs:GetLogEvents"
      ],
      "Resource": "arn:aws:logs:*:*:log-group:/aws/mwaa-serverless/*"
    }
  ]
}'

Step 2: Write the Python module

Create a file called csv_to_json.py with the conversion logic:

# csv_to_json.py
import csv
import json
import boto3
import io

def convert(**kwargs):
    """Read a CSV from S3 and write it back as JSON lines."""
    bucket = "amzn-s3-demo-mwaa-data"
    source_key = "raw/sales_data.csv"
    output_key = "processed/sales_data.json"

    s3 = boto3.client("s3")

    # Read source file
    response = s3.get_object(Bucket=bucket, Key=source_key)
    content = response["Body"].read().decode("utf-8")

    # Parse CSV
    reader = csv.DictReader(io.StringIO(content))
    rows = list(reader)

    # Type inference - convert numeric fields
    for row in rows:
        for key, value in row.items():
            try:
                row[key] = float(value)
            except (ValueError, TypeError):
                pass

    # Write as JSON lines
    output = "\n".join(json.dumps(row) for row in rows) + "\n"
    s3.put_object(Bucket=bucket, Key=output_key, Body=output.encode("utf-8"))

    print(f"Converted {len(rows)} rows to JSON lines")
    print(f"Output: s3://amzn-s3-demo-mwaa-data/{output_key}")
    return {"rows": len(rows), "output_key": output_key}

This function uses boto3 (which comes pre-installed with the MWAA Serverless execution environment) and Python’s built-in csv and json modules. The conversion reads the CSV, infers numeric types, and writes a JSON lines file back to the S3 bucket.

Step 3: Write the verification script

Create a file called verify_output.sh. This script validates the pipeline output by comparing the record count in the source CSV against the output JSON file. If the counts do not match, the task fails with a non-zero exit code, which causes the workflow run to fail.

#!/bin/bash
echo "=== Data Validation ==="

# Count source records (skip CSV header)
SOURCE_COUNT=$(python3 -m awscli s3 cp s3://amzn-s3-demo-mwaa-data/raw/sales_data.csv - | tail -n +2 | wc -l)
echo "Source CSV records: $SOURCE_COUNT"

# Count output records
OUTPUT_COUNT=$(python3 -m awscli s3 cp s3://amzn-s3-demo-mwaa-data/processed/sales_data.json - | wc -l)
echo "Output JSON records: $OUTPUT_COUNT"

# Validate counts match
if [ "$SOURCE_COUNT" -ne "$OUTPUT_COUNT" ]; then
    echo "FAILED: Record count mismatch (source=$SOURCE_COUNT, output=$OUTPUT_COUNT)"
    exit 1
fi

echo "PASSED: Record counts match ($OUTPUT_COUNT records)"
echo "Timestamp: $(date -u +%Y-%m-%dT%H:%M:%SZ)"

This script runs the AWS CLI, which is bundled as a dependency in the code package. The s3 cp streams the file content to stdout without writing to disk, allowing standard shell tools like wc -l and tail to process it. The execution role credentials are automatically available in the execution environment, so the CLI can access S3 without additional configuration.

Step 4: Package and upload the code to Amazon S3

Since the verification script uses the AWS CLI, bundle it as a dependency in the ZIP archive along with your Python module and shell script:

BUCKET="amzn-s3-demo-mwaa-data"
REGION="us-east-1"

# Install awscli into a package directory
pip install awscli \
  --target my_package/ \
  --platform manylinux2014_x86_64 \
  --python-version 3.12 \
  --only-binary=:all:

# Add your module
cp csv_to_json.py my_package/
cp verify_output.sh my_package/

# Create the ZIP archive
cd my_package && zip -r ../code_bundle.zip . && cd ..
# Upload to S3
aws s3 cp code_bundle.zip s3://$BUCKET/code/code_bundle.zip --region $REGION

Upload a sample CSV file for testing:

cat > sales_data.csv << 'EOF'
date,region,product,units,revenue
2026-07-01,us-east,widget-a,150,4500.00
2026-07-01,eu-west,widget-b,89,2670.00
2026-07-02,us-east,widget-a,203,6090.00
2026-07-02,ap-south,widget-c,67,1340.00
2026-07-03,us-east,widget-b,178,5340.00
EOF

aws s3 cp sales_data.csv s3://$BUCKET/raw/sales_data.csv --region $REGION

Step 5: Define the DAG (YAML)

MWAA Serverless uses a declarative YAML format for DAG definitions. Create a file called conversion_dag.yaml:

csv_to_json_pipeline:
  start_date: "2026-01-01"
  schedule: null
  tasks:
    convert_to_json:
      operator: airflow.operators.python.PythonOperator
      python_callable: csv_to_json.convert
    verify_output:
      operator: airflow.operators.bash.BashOperator
      bash_command: "verify_output.sh"
      dependencies:
        - convert_to_json

This DAG defines two tasks:

• convert_to_json – Runs the convert function from the Python module to transform CSV to JSON lines.
• verify_output – Runs a shell script that validates the pipeline output by comparing source and output record counts, failing the task if they do not match.

Upload the DAG definition to S3. Note: You can also run inline Bash commands directly without a shell script.

aws s3 cp conversion_dag.yaml s3://$BUCKET/dags/conversion_dag.yaml --region $REGION

Step 6: Create the workflow

Create the MWAA Serverless workflow, referencing the DAG definition and the code bundle:

ROLE_ARN="arn:aws:iam::<your-account-id>:role/MWAAServerlessExecutionRole"

aws mwaa-serverless create-workflow \
  --name csv-to-json-workflow \
  --definition-s3-location Bucket="$BUCKET",ObjectKey="dags/conversion_dag.yaml" \
  --code '{"S3Location": {"Bucket":"'"$BUCKET"'","ObjectKey":"code/code_bundle.zip"}}' \
  --role-arn $ROLE_ARN \
  --region $REGION

The response includes a WorkflowArn that you use to trigger runs:

{
  "WorkflowArn": "arn:aws:airflow-serverless:us-east-1:123456789012:workflow/csv-to-json-workflow-abc123",
  "CreatedAt": "2026-07-15T10:30:00.000000+00:00",
  "WorkflowVersion": "a1b2c3d4e5f6"
}

Step 7: Run the workflow

Trigger a workflow run:

WORKFLOW_ARN="arn:aws:airflow-serverless:us-east-1:123456789012:workflow/csv-to-json-workflow-abc123"

aws mwaa-serverless start-workflow-run \
  --workflow-arn $WORKFLOW_ARN \
  --region $REGION

The response confirms the run has started:

{
  "RunId": "6OZV9ABF9enHKXk",
  "Status": "STARTING"
}

Step 8: Monitor execution

Check the status of your run:

RUN_ID="6OZV9ABF9enHKXk"

aws mwaa-serverless get-workflow-run \
  --workflow-arn $WORKFLOW_ARN \
  --run-id $RUN_ID \
  --region $REGION

A successful run returns:

{
  "RunDetail": {
    "Duration": 45,
    "RunState": "SUCCESS",
    "TaskInstances": ["ex_abc123_convert_to_json_1", "ex_abc123_verify_output_1"]
  },
  "RunId": "6OZV9ABF9enHKXk",
  "RunType": "ON_DEMAND",
  "WorkflowArn": "arn:aws:airflow-serverless:us-east-1:123456789012:workflow/csv-to-json-workflow-abc123",
  "WorkflowVersion": "a1b2c3d4e5f6"
}

Step 9: Verify the output

Confirm the JSON file was written to the S3 bucket:

# List the output file
aws s3 ls s3://$BUCKET/processed/sales_data.json --region $REGION

You should see the JSON file:

2026-07-15 10:32:45 1847 sales_data.json

You can also verify task-level output in Amazon CloudWatch Logs. Open the log group for your workflow and find the convert_to_json task log stream:

Converted 5 rows to JSON lines
Output: s3://amzn-s3-demo-mwaa-data/processed/sales_data.json

Considerations and limits

When planning your workloads on MWAA Serverless with these operators, keep the following considerations in mind:

• Code bundle size – ZIP archives must be under 250 MB per bundle.
• Network access – Python and Bash tasks do not have internet access. They can reach a limited set of AWS services required for the runtime to function (Amazon S3, Amazon ECR, and Amazon CloudWatch) but cannot call other AWS services or external endpoints. If your workflow requires calls to external APIs, preprocess that data and store it in an Amazon S3 bucket before invoking the workflow.
• Runtime dependencies – boto3 and the Python standard library are pre-installed. For additional packages (such as pandas or requests), bundle them in your ZIP archive following the Amazon MWAA Serverless packaging guidelines.
• Execution timeout – Tasks are subject to the workflow’s configured timeout limits.
• Python version – Check the Amazon MWAA Serverless documentation for the currently supported Python runtime version.
• DAG format – MWAA Serverless uses YAML-based DAG definitions, not traditional Python DAG files. If you are migrating from MWAA Provisioned, you will need to convert your DAGs to the YAML format.
• Operators not supported – Some Airflow community operators and custom plugins are not available in the Serverless runtime. Refer to the documentation for the full compatibility list.

Clean up

To avoid ongoing charges, delete the resources you created in this walkthrough. The following commands remove the workflow, S3 objects, and IAM role:

Note: $WORKFLOW_ARN is defined in Step 7.

# Delete the workflow
aws mwaa-serverless delete-workflow \
  --workflow-arn $WORKFLOW_ARN \
  --region $REGION

Note: $BUCKET is exported in Step 4. If appropriate, delete the bucket as well.

# Remove S3 objects
aws s3 rm s3://$BUCKET/code/code_bundle.zip
aws s3 rm s3://$BUCKET/dags/conversion_dag.yaml
aws s3 rm s3://$BUCKET/raw/sales_data.csv
aws s3 rm s3://$BUCKET/processed/sales_data.json

# Delete the IAM role
aws iam delete-role-policy \
  --role-name MWAAServerlessExecutionRole \
  --policy-name MWAAServerlessAccessPolicy

aws iam delete-role --role-name MWAAServerlessExecutionRole

Conclusion

With native support for PythonOperator and BashOperator, you can now run the custom code execution patterns that many data engineering teams rely on daily directly in MWAA Serverless. Run data transformations, format conversions, validations, and shell scripts in the serverless runtime – without provisioning additional compute or managing containers.

If you are running Airflow workloads on MWAA Provisioned or self-managed infrastructure, your existing PythonOperator and BashOperator logic requires minimal changes. Convert your Python DAG files to the YAML format, package your code as a bundle, and you are ready to run on MWAA Serverless.

To get started, visit the Amazon MWAA Serverless documentation and try the walkthrough earlier in this post with your own data. For pricing details, visit the Amazon MWAA pricing page. We look forward to your feedback.

About the authors