如何从我的Python Spark脚本登录

Question

我有一个Python Spark程序，我用spark-submit运行。我想把日志语句放在其中。

logging.info("This is an informative message.")
logging.debug("This is a debug message.")

我想使用Spark正在使用的相同记录器，以便日志消息以相同的格式出现，并且级别由相同的配置文件控制。我该怎么做呢？

我已经尝试将logging语句放在代码中并从logging.getLogger()开始。在这两种情况下，我都看到Spark的日志消息，但不是我的。我一直在看Python logging documentation，但还是没能从那里弄明白。

不确定这是否是提交给Spark的脚本特有的，或者我不了解日志记录的工作原理。

Answer 1

您可以从SparkContext对象获取记录器：

log4jLogger = sc._jvm.org.apache.log4j
LOGGER = log4jLogger.LogManager.getLogger(__name__)
LOGGER.info("pyspark script logger initialized")

Answer 2

您需要获取spark本身的记录器，默认情况下，getLogger()将为您自己的模块返回记录器。尝试类似的东西：

logger = logging.getLogger('py4j')
logger.info("My test info statement")

它也可能是'pyspark'而不是'py4j'。

如果您在spark程序中使用的函数（以及执行某些日志记录）在与main函数相同的模块中定义，则会产生一些序列化错误。

这是解释here和同一个人的例子给予here

我还在spark 1.3.1上测试了这个

编辑：

要将记录从STDERR更改为STDOUT，您必须删除当前的StreamHandler并添加一个新的。

找到现有的流处理程序（完成后可以删除此行）

print(logger.handlers)
# will look like [<logging.StreamHandler object at 0x7fd8f4b00208>]

可能只有一个，但如果不是，你将不得不更新位置。

logger.removeHandler(logger.handlers[0])

为sys.stdout添加新的处理程序

import sys # Put at top if not already there
sh = logging.StreamHandler(sys.stdout)
sh.setLevel(logging.DEBUG)
logger.addHandler(sh)

Answer 3

交互pyspark和java log4j的关键是jvm。下面是python代码，conf缺少url，但这是关于日志记录。

from pyspark.conf import SparkConf
from pyspark.sql import SparkSession

my_jars = os.environ.get("SPARK_HOME")
myconf = SparkConf()
myconf.setMaster("local").setAppName("DB2_Test")
myconf.set("spark.jars","%s/jars/log4j-1.2.17.jar" % my_jars)
spark = SparkSession\
 .builder\
 .appName("DB2_Test")\
 .config(conf = myconf) \
 .getOrCreate()


Logger= spark._jvm.org.apache.log4j.Logger
mylogger = Logger.getLogger(__name__)
mylogger.error("some error trace")
mylogger.info("some info trace")

Answer 4

我们需要从执行程序登录，而不是从驱动程序节点登录。所以我们做了以下事情：

1. 我们在所有节点上创建了一个/etc/rsyslog.d/spark.conf（使用带有Amazon Elastic Map Reduceso that the Core nodes forwarded sysloglocal1`消息的Bootstrap方法到主节点。
2. 在主节点上，我们启用了UDP和TCP syslog侦听器，并进行了设置，以便将所有local消息记录到/var/log/local1.log。
3. 我们在map函数中创建了一个Python logging模块Syslog记录器。
4. 现在我们可以用logging.info()登录。 ...

我们发现的一件事是在多个执行器上同时处理相同的分区。显然，当它有额外的资源时，Spark会一直这样做。这处理了执行者神秘延迟或失败的情况。

登录map函数已经教会了我们很多Spark如何工作。

Answer 5

在我的情况下，我很高兴将我的日志消息添加到worker stderr，以及通常的spark日志消息。

如果这符合您的需求，那么诀窍是将特定的Python记录器重定向到stderr。

例如，以下灵感来自this answer，对我来说很好：

def getlogger(name, level=logging.INFO):
    import logging
    import sys

    logger = logging.getLogger(name)
    logger.setLevel(level)
    if logger.handlers:
        # or else, as I found out, we keep adding handlers and duplicate messages
        pass
    else:
        ch = logging.StreamHandler(sys.stderr)
        ch.setLevel(level)
        formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
        ch.setFormatter(formatter)
        logger.addHandler(ch)
    return logger

用法：

def tst_log():
    logger = getlogger('my-worker')
    logger.debug('a')
    logger.info('b')
    logger.warning('c')
    logger.error('d')
    logger.critical('e')
    ...

输出（加上上下文的几条线）：

17/05/03 03:25:32 INFO MemoryStore: Block broadcast_24 stored as values in memory (estimated size 5.8 KB, free 319.2 MB)
2017-05-03 03:25:32,849 - my-worker - INFO - b
2017-05-03 03:25:32,849 - my-worker - WARNING - c
2017-05-03 03:25:32,849 - my-worker - ERROR - d
2017-05-03 03:25:32,849 - my-worker - CRITICAL - e
17/05/03 03:25:32 INFO PythonRunner: Times: total = 2, boot = -40969, init = 40971, finish = 0
17/05/03 03:25:32 INFO Executor: Finished task 7.0 in stage 20.0 (TID 213). 2109 bytes result sent to driver