前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >spark-submit --files

spark-submit --files

作者头像
挽风
发布2023-10-17 15:46:40
3560
发布2023-10-17 15:46:40
举报
文章被收录于专栏:小道小道

一、原理

  spark-submit --files通常用来加载外部资源文件,在driver和executor进程中进行访问

  –files和–jars基本相同

二、使用步骤

2.1 添加文件

spark-submit --files file_paths

  其中file_paths可为多种方式:file: | hdfs:// | http:// | ftp:// | local:(多个路径用逗号隔开

代码语言:javascript
复制
spark-submit \
--master yarn \
--deploy-mode cluster \
--principal xxx.com \
--keytab /xxx/keytabs/xxx.keytab \
--driver-java-options "-Dspring.profiles.active=prod -Dorg.springframework.boot.logging.LoggingSystem=none -Djava.ext.dirs=/xxx/CDH-x.x.x-1.cdhx.x.x.p0.xxx/lib/spark/jars/gson-2.8.1.jar,/xxx/CDH-x.x.x-1.cdhx.1.1.p0.xxx/lib/hive/lib/* -Dspark.yarn.dist.files=/xxx/CDH-x.1.1-1.cdhx.1.1.p0.xxx/etc/hadoop/conf.dist/yarn-site.xml" \
--driver-class-path "/xxx/CDH-x.x.x-x.cdhx.x.x.p1000.xxx/jars/gson-2.8.1.jar:$JAVA_HOME/jre/lib/ext/*:/xxx/CDH-x.x.x-1.cdhx.x.x.p1000.xxx/jars/commons-cli-1.4.jar" \
--driver-cores $dc \
--driver-memory $dm \
--num-executors $ne \
--executor-cores $ec \
--executor-memory $em \
--conf spark.sql.crossJoin.enabled=true \
--conf spark.dynamicAllocation.enabled=false \
--conf spark.yarn.maxAppAttempts=1 \
--conf spark.driver.maxResultSize=20G \
--conf spark.sql.shuffle.partitions=$partitions \
--conf spark.default.parallelism=$partitions \
--conf spark.sql.adaptive.enabled=true \
--conf spark.sql.adaptive.shuffle.targetPostShuffleInputSize=128000000 \
--conf spark.yarn.queue=xxx \
--conf spark.shuffle.io.maxRetries=200 \
--conf spark.shuffle.io.retryWait=30 \
--conf spark.port.maxRetries=120 \
--conf spark.core.connection.ack.wait.timeout=6000 \
--conf spark.shuffle.sort.bypassMergeThreshold=300 \
--conf spark.hadoop.hive.exec.dynamic.partition=$dp \
--conf spark.hadoop.hive.exec.dynamic.partition.mode=nonstrict \
--conf spark.hadoop.hive.exec.max.dynamic.partitions=2000 \
--name $name \
--files "$files" \ #/path/服务器本地文件
--class xxxApplication /xxx/xxx-1.0-SNAPSHOT.jar -jn $obj -sq "$sql" -ptby $ptby

2.2 获取文件

2.2.1 方案一
代码语言:javascript
复制
//If you add your external files using "spark-submit --files" your files will be uploaded to this HDFS folder: hdfs://your-cluster/user/your-user/.sparkStaging/application_1449220589084_0508

//application_1449220589084_0508 is an example of yarn application ID!

//1. find the spark staging directory by below code: (but you need to have the hdfs uri and your username)

System.getenv("SPARK_YARN_STAGING_DIR"); --> .sparkStaging/application_1449220589084_0508

//2. find the complete comma separated file paths by using:

System.getenv("SPARK_YARN_CACHE_FILES"); -->
hdfs://yourcluster/user/hdfs/.sparkStaging/application_1449220589084_0508/spark-assembly-1.4.1.2.3.2.0-2950-hadoop2.7.1.2.3.2.0-2950.jar#__spark__.jar,
hdfs://yourcluster/user/hdfs/.sparkStaging/application_1449220589084_0508/your-spark-job.jar#__app__.jar,
hdfs://yourcluster/user/hdfs/.sparkStaging/application_1449220589084_0508/test_file.txt#test_file.txt

   --files会把文件上传到hdfs的.sparkStagin/applicationId目录下。

spark.read().textFile(System.getenv("SPARK_YARN_STAGING_DIR") + "/xxx.xxx")

  textFile不指定hdfs、file或者其他前缀的话默认是hdfs://yourcluster/user/your_username下的相对路径。

2.2.2 方案二 SparkFiles.get(fileName)

SparkFiles.get(fileName) 适用于local模式

代码语言:javascript
复制
JavaRDD<String> stringJavaRDD = sparkcontext.textFile(SparkFiles.get(fileName));

List<String> collect = stringJavaRDD.collect();

[注意事项]

  在cluster模式下(-- deploy-mode cluster ),-- files必须使用全局可视的地址(比如hdfs),否则driver将无法找到文件,出现FileNotFoundException。这是因为driver会在集群中任意一台worker节点上运行,使用本地地址无法找到文件。FileNotFoundException异常出现在SparkSession的getOrCreate()初始化方法中,因为此方法会调用addFile(),但是确找不到文件,导致SparkSession初始化失败。注意:–jars原理相同,但是getOrCreate()中调用addJars出现异常,但是并不会导SparkSession初始化失败,程序会继续运行。

  值得一提的是,在cluster模式下,spark-submit   --deploy-mode cluster   path-to-jar,其中path-to-jar也必须是全局可视路径,否则会发生找不到jar的异常。

2.2.3 方案三 new FileInputStream(fileName)
代码语言:javascript
复制
FileInputStream sqlstream = new FileInputStream(fileName);

StringBuilder sqlContent = new StringBuilder();

Scanner scanner = new Scanner(sqlstream, "UTF-8");

while (scanner.hasNextLine()) {
	String line = scanner.nextLine();
    sqlContent.append(line).append("\n");
}

适用于local、yarn client、yarn cluster模式,

2.2.4 方案四
代码语言:javascript
复制
Properties properties = new Properties();
properties.load(Thread.currentThread().getContextClassLoader().getResourceAsStream("test.properties"));

适应于yarn、cluster模式

本文参与 腾讯云自媒体分享计划,分享自作者个人站点/博客。
原始发表:2023-10-11,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体分享计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 一、原理
  • 二、使用步骤
    • 2.1 添加文件
      • 2.2 获取文件
        • 2.2.1 方案一
        • 2.2.2 方案二 SparkFiles.get(fileName)
        • 2.2.3 方案三 new FileInputStream(fileName)
        • 2.2.4 方案四
    相关产品与服务
    专用宿主机
    专用宿主机(CVM Dedicated Host,CDH)提供用户独享的物理服务器资源,满足您资源独享、资源物理隔离、安全、合规需求。专用宿主机搭载了腾讯云虚拟化系统,购买之后,您可在其上灵活创建、管理多个自定义规格的云服务器实例,自主规划物理资源的使用。
    领券
    问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档