Pyspark: Prevent removal of terminal ansi-escape characters in logging while using pyspark
<p>I have put together a custom formatter for a logger and I am using pyspark, but it looks like all of my color is removed on the command-line.  I can confirm that the escape sequences are present within the record of each emitted value, but it appears that they're stripped when sent to the terminal.  </p>

<p>Why?</p>

<pre class="lang-py prettyprint-override"><code>import datetime
import logging

import colorama
from pygments import highlight
from pygments.lexers import JsonLexer
from pygments.formatters import Terminal256Formatter

# Required for colored output
colorama.init()


class CustomFormatter(logging.Formatter):
    '''Modifies the level prefix of the log with the following level
    information:

    !!! - critical
     !  - error
     ?  - warn
        - info
     -  - debug
    '''
    default_prefix = '???'  # used with non-generic levels

    color_mapping = {
        logging.CRITICAL: colorama.Fore.RED + colorama.Style.BRIGHT,
        logging.ERROR: colorama.Fore.RED + colorama.Style.BRIGHT,
        logging.WARNING: colorama.Fore.YELLOW + colorama.Style.BRIGHT,
        logging.DEBUG: colorama.Style.DIM,
    }

    prefix_mapping = {
        logging.CRITICAL: '!!!',
        logging.ERROR: ' ! ',
        logging.WARNING: ' ? ',
        logging.INFO: '   ',
        logging.DEBUG: ' Â· ',
    }

    def format(self, record):
        # Capture relevant record data
        level = self.prefix_mapping.get(record.levelno) or self.default_prefix
        msecs = datetime.datetime.fromtimestamp(record.created).strftime("%Y-%m-%d %H:%M:%S")
        msg = record.msg.rstrip('\n')

        # Setup colors
        color = self.color_mapping.get(record.levelno) or ''
        dim = colorama.Style.DIM
        reset = colorama.Fore.RESET + colorama.Style.RESET_ALL
        name = record.name
        func = record.funcName

        # Setup output
        lexer = JsonLexer()
        formatter = Terminal256Formatter()
        try:
            msg = '\n'.join(
                highlight(m, lexer, formatter).rstrip('\n')
                for m in msg.split('\n')
            )
        except:
            pass
        data = {k: v for k, v in locals().items()}
        d = '{color}{level}{reset} {dim}{msecs} [{name}]{reset} {msg}'.format(**data)
        record.msg = d

        # Dump
        return super(CustomFormatter, self).format(record)
</code></pre>

<p>Usage:</p>

<pre class="lang-py prettyprint-override"><code>import logging
from  CustomFormatter import CustomFormatter

def get_logger(name, level=None):
    level = logging.DEBUG if not isinstance(level, int) else level
    handler = logging.StreamHandler(sys.stdout)
    handler.level = level or logging.INFO
    formatter = CustomFormatter()
    handler.setFormatter(formatter)
    logger = logging.getLogger(name)
    logger.addHandler(handler)
    return logger

logger = get_logger('tester')
logger.error('Error here')
</code></pre>

<p>I spent some time digging into this, and I found that the terminal escape sequences are different when loading under pyspark.  The way I fixed it is by using pygments to run over the terminal output I created (see function:  <code>fix_for_spark</code>).</p>

<pre><code># -*- coding: utf-8 -*-
import datetime
import json
import logging
import os

import colorama
from pygments import highlight
from pygments.lexers import JsonLexer
from pygments.formatters import Terminal256Formatter

# Required for colored output
colorama.init()


class CustomFormatter(logging.Formatter):

    '''Modifies the level prefix of the log with the following level
    information:

    !!! - critical
     !  - error
     ?  - warn
        - info
     -  - debug
    '''
    default_prefix = '???'  # used with non-generic levels

    PYGMENTS_LEXER = JsonLexer()
    PYGMENTS_FORMATTER = Terminal256Formatter()

    color_mapping = {
        logging.CRITICAL: colorama.Fore.RED + colorama.Style.BRIGHT,
        logging.ERROR: colorama.Fore.RED + colorama.Style.BRIGHT,
        logging.WARNING: colorama.Fore.YELLOW + colorama.Style.BRIGHT,
        logging.DEBUG: colorama.Style.DIM,
    }

    prefix_mapping = {

        logging.CRITICAL: '!!!',
        logging.ERROR: ' ! ',
        logging.WARNING: ' ? ',
        logging.INFO: '  ï¸ ',
        logging.DEBUG: ' Â· ',
    }

    def fix_for_spark(self, string):
        if os.environ.get('SPARK_ENV_LOADED'):
            # Setup output
            new_string = []
            for s in string.split('\n'):
                s = highlight(string, self.PYGMENTS_LEXER, self.PYGMENTS_FORMATTER)
                new_string.append(s.rstrip('\n'))
            string = '\n'.join(new_string)
        return string

    def format(self, record):
        # Capture relevant record data
        data = dict(
            level=self.prefix_mapping.get(record.levelno) or self.default_prefix,
            msecs=datetime.datetime.fromtimestamp(record.created).strftime("%Y-%m-%d %H:%M:%S"),

            # Setup colors
            color=self.color_mapping.get(record.levelno) or '',
            dim=colorama.Style.DIM,
            reset=colorama.Fore.RESET + colorama.Style.RESET_ALL,
            name=record.name,
            func=record.funcName,
        )

        # Format msg
        prefix = '{color}{level}{reset} {dim}{msecs}{reset} {color}[{name}]{reset}'
        prefix = prefix.format(**data)
        prefix = self.fix_for_spark(prefix)
        msg = record.msg
        if not isinstance(msg, str):
            try:
                msg = json.dumps(msg, indent=4, sort_keys=True)
            except:
                msg = str(msg)
        dmsg = []
        for m in msg.split('\n'):
            m = highlight(m, self.PYGMENTS_LEXER, self.PYGMENTS_FORMATTER).rstrip('\n')
            m = self.fix_for_spark(m)
            dmsg.append(m)
        dmsg = '\n'.join(dmsg)
        data.update(locals().items())
        template = prefix + ' {msg}'
        record.msg = '\n'.join(template.format(msg=m) for m in dmsg.split('\n'))
        # Dump
        return super(CustomFormatter, self).format(record)
</code></pre>

python logging pyspark ansi-escape