如何将任何新的库（如spack-sftp）添加到我的Pyspark代码中？

Question

当我试图在spark配置中设置包依赖项“spark sftp”时，我得到ClassNotFoundException。但当我使用以下命令执行脚本时，它会起作用：

spark submit——包com。springml:spark-sftp_2.11:1.1.1测试。py

下面是我的代码。有人能告诉我，如何在不将包作为参数传递给spark submit的情况下执行pyspark脚本吗？

import sys
import datetime
import pyspark
from pyspark.sql import *
from pyspark.sql import SparkSession, SQLContext, Row, HiveContext
from pyspark import SparkContext

#Create new config
conf = (pyspark.conf.SparkConf()
.set("spark.driver.maxResultSize", "16g")
.set("spark.driver.memory", "20g")
.set("spark.executor.memory", "20g")
.set("spark.executor.cores", "5")
.set("spark.shuffle.service.enabled", "true")
.set("spark.dynamicAllocation.enabled", "true")
.set("spark.dynamicAllocation.initialExecutors", "24")
.set("spark.dynamicAllocation.minExecutors", "6")
.set("spark.submit.deployMode", "client")
.set("spark.jars.packages", "com.springml:spark-sftp_2.11:1.1.1")
.set("spark.python.worker.memory", "4g")
.set("spark.default.parallelism", "960")
.set("spark.executor.memoryOverhead", "4g")
.setMaster("yarn-client"))


# Create new context
spark = SparkSession.builder.appName("AppName").config(conf=conf).enableHiveSupport().getOrCreate()
spark.sparkContext.setLogLevel("WARN")


df = spark.read.format("com.springml.spark.sftp").option("host", "HOST").option("username", "HOSTNAME").option("password", "pass").option("fileType", "csv").option("inferSchema", "true").load("/test/sample.csv")

输出java.lang.reflect.ethod.invokeClassNotFoundException：未能找到数据源：ethod.java:498请在py4j.reflection.找到包nvoker.invokeDataSource$. lookupDataSource（DataSnvoker.java:244）在py4j.reflection.DataFrameRngine.invoke（DataFrameRngine.java:357）在ateway.invokeDataFrameRateway.java:282（DataFrameRpy4j.commands.）在ommand.invokeNativeMethodAccessorImpl.invoke0（Native Method）在sun.reflect.NativeMethodAccessorImpl.invoke（NativeMethodAccessorImpl.java:62）在sun.reflect.在:: Mjava.lang.（Mcom.springml.spark.sftp.）在http://spark.apache.org/third-party-projects.htmlMethodIorg.apache.spark.sql.execution.datasources.（MethodIource.scala:635）在org.apache.spark.sql.反射Eeader.load（反射Eeader.scala:190）在py4j. Gorg.apache.spark.sql.（Geader.load）在eader.scala:174AbstractCsun.reflect.方法（AbstractCommand. java： 132）在py4j. CallCommand.执行（CallCommand. java： 79）在py4j. GatewayConnec. run（GatewayConnec. java： 238）在jjj.

Answer 1

提交spark作业时，可以指定要安装的软件包。对于这个，您可以将这个maven依赖项指定为：

> $SPARK_HOME/bin/spark-shell --packages com.springml:spark-sftp_2.11:1.1.3