Я пытаюсь запустить искровой код, который будет читать таблицу в базе данных postgres и вставлять ее в таблицу Hive на HDFS.Для этого я установил свойства соединения в файле свойств, как показано ниже:
devUserName=usrname
devPassword=pwd
gpDriverClass=org.postgresql.Driver
gpDevUrl=jdbc:postgresql://ip:port/dbname?ssl=true&sslfactory=org.postgresql.ssl.NonValidatingFactory
И считал файл свойств следующим образом:
val conFile = "testconnection.properties"
val properties = new Properties()
properties.load(new FileInputStream(conFile))
// GP Properties
val connectionUrl = properties.getProperty("gpDevUrl")
val devUserName = properties.getProperty("devUserName")
val devPassword = properties.getProperty("devPassword")
val driverClass = properties.getProperty("gpDriverClass")
Я отправляю файл jar в следующемпуть:
SPARK_MAJOR_VERSION=2 spark-submit --conf spark.ui.port=4090 --driver-class-path /home/usrname/jars/postgresql-42.1.4.jar --master=yarn --deploy-mode=cluster --driver-memory 40g --driver-cores 3 --executor-memory 40g --executor-cores 20 --num-executors 5 --files /usr/hdp/current/spark2-client/conf/hive-site.xml,testconnection.properties --jars /home/usrname/jars/postgresql-42.1.4.jar --class com.partition.source.YearPartition splinter_2.11-0.1.jar --keytab /home/usrname/usrname.keytab --principal usrname@DEV.COM --name Splinter --conf spark.executor.extraClassPath=/home/usrname/jars/postgresql-42.1.4.jar
Сбой задания со следующим сообщением об ошибке:
client token: Token { kind: YARN_CLIENT_TOKEN, service: }
diagnostics: User class threw exception: java.sql.SQLException: No suitable driver
ApplicationMaster host: 10.230.137.190
ApplicationMaster RPC port: 0
queue: default
start time: 1537545547309
final status: FAILED
tracking URL: http://ip:port/proxy/application_123456789_76/
user: fdlhdpetl
18/09/21 15:59:38 INFO Client: Deleted staging directory hdfs://dev/user/usrname/.sparkStaging/application_123456789_76
Exception in thread "main" org.apache.spark.SparkException: Application application_123456789_76 finished with failed status
at org.apache.spark.deploy.yarn.Client.run(Client.scala:1187)
at org.apache.spark.deploy.yarn.Client$.main(Client.scala:1233)
at org.apache.spark.deploy.yarn.Client.main(Client.scala)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:782)
at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:180)
at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:205)
at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:119)
at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
18/09/21 15:59:38 INFO ShutdownHookManager: Shutdown hook called
18/09/21 15:59:38 INFO ShutdownHookManager: Deleting directory /tmp/spark-59934097-dfe9-4ad2-a66a-ff93d42cf838
Мои зависимости в файле build.sbt:
name := "Splinter"
version := "0.1"
scalaVersion := "2.11.8"
libraryDependencies ++= Seq(
"org.apache.spark" %% "spark-core" % "2.0.0" % "provided",
"org.apache.spark" %% "spark-sql" % "2.0.0" % "provided",
"org.json4s" %% "json4s-jackson" % "3.2.11" % "provided",
"org.apache.httpcomponents" % "httpclient" % "4.5.3"
)
// https://mvnrepository.com/artifact/org.postgresql/postgresql
libraryDependencies += "org.postgresql" % "postgresql" % "42.1.4"
Код, используемый дляпрочитайте таблицу на postgres:
def prepareFinalDF(splitColumns:List[String], textList: ListBuffer[String], allColumns:String, dataMapper:Map[String, String], partition_columns:Array[String], spark:SparkSession): DataFrame = {
val execQuery = s"select ${allColumns}, 0 as ${flagCol} from schema.tablename where period_year='2017'"
val yearDF = spark.read.format("jdbc").option("url", connectionUrl).option("dbtable", s"(${execQuery}) as year2017").option("user", devUserName).option("password", devPassword).option("numPartitions",20).load()
val totalCols:List[String] = splitColumns ++ textList
val cdt = new ChangeDataTypes(totalCols, dataMapper)
hiveDataTypes = cdt.gpDetails()
prepareHiveTableSchema(hiveDataTypes, partition_columns)
val allColsOrdered = yearDF.columns.diff(partition_columns) ++ partition_columns
val allCols = allColsOrdered.map(colname => org.apache.spark.sql.functions.col(colname))
val resultDF = yearDF.select(allCols:_*)
val stringColumns = resultDF.schema.fields.filter(x => x.dataType == StringType).map(s => s.name)
val finalDF = stringColumns.foldLeft(resultDF) {
(tempDF, colName) => tempDF.withColumn(colName, regexp_replace(regexp_replace(col(colName), "[\r\n]+", " "), "[\t]+"," "))
}
finalDF
}
execQuery содержит: select * from schema.tablename where period_year=2017
val dataDF = prepareFinalDF(splitColumns, textList, allColumns, dataMapper, partition_columns, spark)
dataDF.createOrReplaceTempView("preparedDF")
spark.sql("set hive.exec.dynamic.partition.mode=nonstrict")
spark.sql("set hive.exec.dynamic.partition=true")
spark.sql(s"INSERT OVERWRITE TABLE default.xx_gl_forecast PARTITION(${prtn_String_columns}) select * from preparedDF")
Я проверил банку, упомянутую в spark-submit, и она присутствует в данной директории, но онапо-прежнему не удается при отправке.Может ли кто-нибудь сказать мне, какую ошибку я здесь делаю?Есть ли какой-то определенный порядок, которому нужно следовать при задании параметров в spark-submit?