CloudWatch Logging

def cloudwatch_logger(
    log_group: str,
    log_stream: Optional[str] = None,
    log_level: str = "INFO",
    **kwargs
) -> LoggerDefinition:
    """
    Logger that sends Dagster logs to CloudWatch.
    
    Parameters:
        log_group: CloudWatch log group name
        log_stream: CloudWatch log stream name (auto-generated if None)
        log_level: Minimum log level to send
        **kwargs: Additional CloudWatch configuration
        
    Returns:
        LoggerDefinition: Configured CloudWatch logger
    """

from dagster import job, op, Definitions
from dagster_aws.cloudwatch import cloudwatch_logger

@op
def log_processing_status(context):
    context.log.info("Starting data processing")
    # Process data...
    context.log.info("Data processing completed successfully")
    return "success"

@job(
    logger_defs={
        "cloudwatch": cloudwatch_logger.configured({
            "log_group": "/dagster/jobs",
            "log_level": "INFO"
        })
    }
)
def monitored_job():
    log_processing_status()

defs = Definitions(jobs=[monitored_job])