读取驱动程序使用 spark-submit 发送的文件

Question

我通过运行发送 Spark 作业以在远程集群上运行

spark-submit ... --deploy-mode cluster --files some.properties ...

我想通过 driver 代码读取some.properties 文件的内容，即在创建 Spark 上下文和启动 RDD 任务之前。该文件会复制到远程驱动程序，但不会复制到驱动程序的工作目录。

我知道的解决这个问题的方法是：

1. 将文件上传到 HDFS
2. 将文件存储在应用 jar 中

两者都不方便，因为这个文件在提交的开发机器上经常更改。

有没有办法读取在驱动程序代码主方法期间使用--files 标志上传的文件？

Answer 1

是的，您可以访问通过 --files 参数上传的文件。

这就是我能够访问通过--files 传入的文件的方式：

./bin/spark-submit \
--class com.MyClass \
--master yarn-cluster \
--files /path/to/some/file.ext \
--jars lib/datanucleus-api-jdo-3.2.6.jar,lib/datanucleus-rdbms-3.2.9.jar,lib/datanucleus-core-3.2.10.jar \
/path/to/app.jar file.ext

在我的 Spark 代码中：

val filename = args(0)
val linecount = Source.fromFile(filename).getLines.size

我确实相信这些文件会下载到与放置 jar 的同一目录中的工作程序中，这就是为什么只需传递文件名而不是 Source.fromFile 的绝对路径即可。

Answer 2

经过调查，我找到了解决上述问题的一种方法。在 spark-submit 期间发送 any.properties 配置，并在 SparkSession 初始化前后由 spark 驱动程序使用。希望对你有帮助。

any.properties

spark.key=value
spark.app.name=MyApp

SparkTest.java

import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;

public class SparkTest{

  public Static void main(String[] args){

    String warehouseLocation = new File("spark-warehouse").getAbsolutePath();

    Config conf = loadConf();
    System.out.println(conf.getString("spark.key"));

    // Initialize SparkContext and use configuration from properties
    SparkConf sparkConf = new SparkConf(true).setAppName(conf.getString("spark.app.name"));

    SparkSession sparkSession = 
    SparkSession.builder().config(sparkConf).config("spark.sql.warehouse.dir", warehouseLocation)
                .enableHiveSupport().getOrCreate();

    JavaSparkContext javaSparkContext = new JavaSparkContext(sparkSession.sparkContext());

  }


  public static Config loadConf() {

      String configFileName = "any.properties";
      System.out.println(configFileName);
      Config configs = ConfigFactory.load(ConfigFactory.parseFile(new java.io.File(configFileName)));
      System.out.println(configs.getString("spark.key")); // get value from properties file
      return configs;
   }
}

Spark 提交：

spark-submit --class SparkTest --master yarn --deploy-mode client --files any.properties,yy-site.xml --jars ...........

Answer 3

--files 和 --archives 选项支持使用 # 指定文件名，就像 Hadoop。

例如，您可以指定：--files localtest.txt#appSees.txt，这会将您在本地命名为 localtest.txt 的文件上传到 Spark 工作目录，但这将通过名称链接到 appSees.txt，并且您的应用程序应使用名称作为appSees.txt 在YARN 上运行时引用它。

这适用于我在 yarn/client 和 yarn/cluster 模式下的 Spark Streaming 应用程序。

Answer 4

使用spark-submit --help，会发现这个选项只适用于executor的工作目录而不是driver。

--files FILES: Comma-separated list of files to be placed in the working directory of each executor.

Answer 5

这是我在 Python Spark 中开发的一个很好的解决方案，用于将任何数据作为文件从外部集成到您的大数据平台。

玩得开心。

# Load from the Spark driver any local text file and return a RDD (really useful in YARN mode to integrate new data at the fly)
# (See https://community.hortonworks.com/questions/38482/loading-local-file-to-apache-spark.html)
def parallelizeTextFileToRDD(sparkContext, localTextFilePath, splitChar):
    localTextFilePath = localTextFilePath.strip(' ')
    if (localTextFilePath.startswith("file://")):
        localTextFilePath = localTextFilePath[7:]
    import subprocess
    dataBytes = subprocess.check_output("cat " + localTextFilePath, shell=True)
    textRDD = sparkContext.parallelize(dataBytes.split(splitChar))
    return textRDD

# Usage example
myRDD = parallelizeTextFileToRDD(sc, '~/myTextFile.txt', '\n') # Load my local file as a RDD
myRDD.saveAsTextFile('/user/foo/myTextFile') # Store my data to HDFS

Answer 6

解决此问题的一种方法是，您可以通过调用SparkContext.getOrCreate() 创建一个临时的SparkContext，然后在SparkFiles.get('FILE') 的帮助下读取您在--files 中传递的文件。

读取文件后，在SparkConf() 变量中检索您需要的所有必要配置。

之后调用这个函数：

SparkContext.stop(SparkContext.getOrCreate())

这将破坏现有的SparkContext，然后在下一行中简单地使用这样的必要配置初始化一个新的SparkContext。

sc = SparkContext(conf=conf).getOrCreate()

你得到了一个SparkContext 并具有所需的设置

Answer 7

在 pyspark 中，我发现轻松实现这一点非常有趣，首先像这样安排你的工作目录：

/path/to/your/workdir/
|--code.py
|--file.txt

然后在你的 code.py 主函数中，像往常一样读取文件：

if __name__ == "__main__":
    content = open("./file.txt").read()

然后不加任何具体配置就提交如下：

spark-submit code.py

它运行正确，这让我很惊讶。我想提交过程将所有文件和子目录文件完全存档，并将它们发送到 pyspark 中的驱动程序，而您应该自己将它们存档在 scala 版本中。顺便说一句，--files 和 --archives 选项都在 worker 而不是驱动程序中工作，这意味着您只能在 RDD 转换或操作中访问这些文件。