如何从Python Spark脚本登录

Question

我有一个Python Spark程序，我使用Spark submit运行该程序。我想把日志记录语句放进去。

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

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

我尝试将日志记录语句放入代码中，并从日志记录开始。getLogger（）。在这两种情况下，我都能看到Spark的日志消息，但我看不到。我一直在看Python日志文档，但没有从中找到答案。

不确定这是否是特定于提交给Spark的脚本，或者只是我不了解日志如何工作。

Answer 1

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

>

我们创建了一个/etc/rsyslog。d/火花。所有节点上的conf（使用Amazon Elastic Map Reduce的引导方法，以便核心节点将sysloglocal1`消息转发到主节点。

在主节点上，我们启用了UDP和TCP系统日志侦听器，并将其设置为所有local消息都记录到/var/log/local1。日志。

我们在map函数中创建了一个Pythonlogging模块Syslog记录器。

现在我们可以使用日志记录进行日志记录。info（）。。。

我们发现，同一分区在多个执行器上同时处理。显然，Spark在拥有额外资源的情况下一直在这样做。这可以处理执行人神秘延迟或失败的情况。

登录map函数让我们了解了很多Spark的工作原理。

Answer 2

您需要为spark本身获取记录器，默认情况下，getLogger（）将为您自己的模块返回记录器。尝试以下方法：

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

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

如果您在spark程序中使用的函数（并执行一些日志记录）与主函数定义在同一个模块中，则会出现一些序列化错误。

这里解释了这一点，并给出了同一个人的示例

我还在火花1.3.1上测试了这个

编辑：

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

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

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

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

logger.removeHandler(logger.handlers[0])

为sys添加新的处理程序。标准输出

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

Answer 3

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

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