我的sparkDF.persist(DISK_ONLY)数据存储在哪里?

maf*_*ffe 5 hadoop persist apache-spark

我想进一步了解hadoop的持久策略。

当我使用DISK_ONLY-strategy持久存储数据帧时,我的数据存储在哪里(路径/文件夹...)?在哪里指定此位置?

ste*_*ino 6

对于简短的回答,我们可以看一下有关以下内容的文档spark.local.dir

用于Spark中“临时”空间的目录,包括映射输出文件和存储在磁盘上的RDD。它应该在系统中的快速本地磁盘上。它也可以是不同磁盘上多个目录的逗号分隔列表。注意:在Spark 1.0和更高版本中,它将由集群管理器设置的SPARK_LOCAL_DIRS(独立,Mesos)或LOCAL_DIRS(YARN)环境变量覆盖。

为了更深入地了解,我们可以看一下代码:a DataFrame(仅是a Dataset[Row])基于RDDs,并且利用了相同的持久性机制。RDDs将此委托给SparkContext,这表示它具有持久性。然后,该任务实际上由org.apache.spark.storage包中的几个类来处理:首先,BlockManagerJust管理要保留的数据块以及如何执行的策略,将实际的持久性委派给DiskStore(当然,在磁盘上写入时)代表用于编写的高级界面,而又具有DiskBlockManager用于更底层操作的。

希望您对现在的位置有所了解,以便我们可以继续了解数据的实际保存位置以及如何配置它:DiskBlockManager调用helper Utils.getConfiguredLocalDirs,出于实用性考虑,我将在此处复制(从链接的2.2.1版本(撰写本文时为最新版本):

def getConfiguredLocalDirs(conf: SparkConf): Array[String] = {
    val shuffleServiceEnabled = conf.getBoolean("spark.shuffle.service.enabled", false)
    if (isRunningInYarnContainer(conf)) {
        // If we are in yarn mode, systems can have different disk layouts so we must set it
        // to what Yarn on this system said was available. Note this assumes that Yarn has
        // created the directories already, and that they are secured so that only the
        // user has access to them.
        getYarnLocalDirs(conf).split(",")
    } else if (conf.getenv("SPARK_EXECUTOR_DIRS") != null) {
        conf.getenv("SPARK_EXECUTOR_DIRS").split(File.pathSeparator)
    } else if (conf.getenv("SPARK_LOCAL_DIRS") != null) {
        conf.getenv("SPARK_LOCAL_DIRS").split(",")
    } else if (conf.getenv("MESOS_DIRECTORY") != null && !shuffleServiceEnabled) {
        // Mesos already creates a directory per Mesos task. Spark should use that directory
        // instead so all temporary files are automatically cleaned up when the Mesos task ends.
        // Note that we don't want this if the shuffle service is enabled because we want to
        // continue to serve shuffle files after the executors that wrote them have already exited.
        Array(conf.getenv("MESOS_DIRECTORY"))
    } else {
        if (conf.getenv("MESOS_DIRECTORY") != null && shuffleServiceEnabled) {
        logInfo("MESOS_DIRECTORY available but not using provided Mesos sandbox because " +
            "spark.shuffle.service.enabled is enabled.")
        }
        // In non-Yarn mode (or for the driver in yarn-client mode), we cannot trust the user
        // configuration to point to a secure directory. So create a subdirectory with restricted
        // permissions under each listed directory.
        conf.get("spark.local.dir", System.getProperty("java.io.tmpdir")).split(",")
    }
}
Run Code Online (Sandbox Code Playgroud)

我认为该代码很容易解释,并且注释很好(并且与文档内容完全匹配):在Yarn上运行时,有一个特定的策略依赖于Yarn容器的存储,在Mesos中,它要么使用Mesos沙箱(除非启用了随机播放服务),在所有其他情况下,它将转到设置为spark.local.dir或的位置java.io.tmpdir(可能是/tmp/)。

因此,如果您只是在玩耍,数据很可能存储在之下/tmp/,否则它在很大程度上取决于您的环境和配置。


maf*_*ffe 4

总结一下我的 YARN 环境:

在 @stefanobaghino 的指导下,我能够在加载纱线配置的代码中更进一步。

val localDirs = Option(conf.getenv("LOCAL_DIRS")).getOrElse("")
Run Code Online (Sandbox Code Playgroud)

这是在yarn-default.xml中的yarn.nodemanager.local-dirs选项中设置的

我的问题的背景是,由错误引起的

2018-01-23 16:57:35,229 WARN org.apache.hadoop.yarn.server.nodemanager.DirectoryCollection: Directory /data/1/yarn/local error, used space above threshold of 98.5%, removing from list of valid directories
Run Code Online (Sandbox Code Playgroud)

我的 Spark 作业有时会被终止,我想了解在运行作业时该磁盘是否也用于我的持久数据(实际上数量很大)。

事实证明,这正是使用磁盘策略保存数据时数据所在的文件夹。

非常感谢您在这个问题上的所有有用指导!