PySpark logging from the executor

by

You cannot use local log4j logger on executors. Python workers spawned by executors jvms has no “callback” connection to the java, they just receive commands. But there is a way to log from executors using standard python logging and capture them by YARN.

On your HDFS place python module file that configures logging once per python worker and proxies logging functions (name it logger.py):

import os
import logging
import sys

class YarnLogger:
    @staticmethod
    def setup_logger():
        if not 'LOG_DIRS' in os.environ:
            sys.stderr.write('Missing LOG_DIRS environment variable, pyspark logging disabled')
            return 

        file = os.environ['LOG_DIRS'].split(',')[0] + '/pyspark.log'
        logging.basicConfig(filename=file, level=logging.INFO, 
                format="%(asctime)s.%(msecs)03d %(levelname)s %(module)s - %(funcName)s: %(message)s")

    def __getattr__(self, key):
        return getattr(logging, key)

YarnLogger.setup_logger()

Then import this module inside your application:

spark.sparkContext.addPyFile('hdfs:///path/to/logger.py')
import logger
logger = logger.YarnLogger()

And you can use in inside your pyspark functions like normal logging library:

def map_sth(s):
    logger.info("Mapping " + str(s))
    return s

spark.range(10).rdd.map(map_sth).count()

The pyspark.log will be visible on resource manager and will be collected on application finish, so you can access these logs later with yarn logs -applicationId .....
enter image description here

More Related Contents:

  1. Calling Java/Scala function from a task
  2. Efficient string matching in Apache Spark
  3. Load CSV file with Spark
  4. How to link PyCharm with PySpark?
  5. How to turn off INFO logging in Spark?
  6. Count number of non-NaN entries in each column of Spark dataframe with Pyspark
  7. importing pyspark in python shell
  8. collect_list by preserving order based on another variable
  9. Rename nested field in spark dataframe
  10. Spark DataFrame: Computing row-wise mean (or any aggregate operation)
  11. How to determine if object is a valid key-value pair in PySpark
  12. PySpark converting a column of type ‘map’ to multiple columns in a dataframe
  13. A list as a key for PySpark’s reduceByKey
  14. Best way to get the max value in a Spark dataframe column
  15. How to map features from the output of a VectorAssembler back to the column names in Spark ML?
  16. Pyspark replace strings in Spark dataframe column
  17. Apache Spark Python Cosine Similarity over DataFrames
  18. Pivot String column on Pyspark Dataframe
  19. Pyspark: Parse a column of json strings
  20. GroupBy column and filter rows with maximum value in Pyspark
  21. Add Jar to standalone pyspark
  22. Spark mllib predicting weird number or NaN
  23. Multiple condition filter on dataframe
  24. How to return a “Tuple type” in a UDF in PySpark?
  25. How to connect HBase and Spark using Python?
  26. How to process RDDs using a Python class?
  27. Pyspark 2.4.0, read avro from kafka with read stream – Python
  28. How to pivot on multiple columns in Spark SQL?
  29. Apache Spark — Assign the result of UDF to multiple dataframe columns
  30. how to add Row id in pySpark dataframes [duplicate]

Leave a Comment