前置

centos7系統(tǒng) spark版本? scala版本匹配

安裝包下載

wget https://downloads.lightbend.com/scala/2.12.6/scala-2.12.6.rpm

wget https://www.apache.org/dyn/closer.lua/spark/spark-2.3.1/spark-2.3.1-bin-hadoop2.7.tgz

備注：spark安裝的機器scala同步安裝

遠程調試Spark的job（local模式）

(1)? 在$SPARK_HOME的conf目錄下的spark-env.sh文件中添加環(huán)境變量：

exportSPARK_SUBMIT_OPTS="-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=5005"

(2)? 在IDEA中添加配置Debug的Remote項：-

agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=5005

(3) 以local模式提交job：

spark-submit --classorg.apache.spark.examples.JavaWordCount? --master local[*] /mnt/share/codes/JavaProjects/SparkDemo/target/SparkDemo-1.0-SNAPSHOT-jar-with-dependencies.jar

(4) 在IDEA中使用（2）中的配置進行Debug，附加參考鏈接

示例

?一個典型的SparkConf的取值示例如下

"spark.app.id"->"local-1559438023886"

"spark.master"->"local[*]"

"spark.driver.port"->"43433"

"spark.executor.id"->"driver"

"spark.app.name"->"Java Spark SQL basic example"

"spark.submit.deployMode"->"client"

"spark.some.config.option"->"some-value"

"spark.jars"->"file:/mnt/share/codes/JavaProjects/SparkDemo/target/SparkDemo-1.0-SNAPSHOT-jar-with-dependencies.jar"

"spark.driver.host"->"gitserver"

讀寫外部數(shù)據(jù)源（JDBC）

Dataset<Row> jdbcDF = spark.read().format("jdbc").option("driver","com.mysql.cj.jdbc.Driver").option("url","jdbc:mysql://XXX:3306/XXX?useSSL=false&serverTimezone=UTC").option("dbtable","project_branches").option("user","XX").option("password","XXX").load();

最新版本Spark不支持com.mysql.jdbc.Driver ，將mysql8.0版本通過com.mysql.cj.jdbc.Driver進行替代

Properties connectionProperties = new Properties();

connectionProperties.put("user","XXXX");

connectionProperties.put("password","XXXXX");

//connectionProperties.put("driver","com.mysql.cj.jdbc.Driver");//這種方法中driver配置不一定得寫進來Dataset<Row> jdbcDF2 = spark.read().jdbc("jdbc:mysql://XXX:3306/tspark?useSSL=false&serverTimezone=UTC","project_branches", connectionProperties);

保存數(shù)據(jù)到JDBC

jdbcDF.write().mode(SaveMode.Append).format("jdbc").option("driver","com.mysql.cj.jdbc.Driver").option("url","jdbc:mysql://XXX:3306/tspark?useSSL=false&serverTimezone=UTC").option("dbtable","project_branches_backup").option("user","XXXX").option("password","XXXX").save();

jdbcDF2.write().mode(SaveMode.Append).jdbc("jdbc:mysql://XXX:3306/tspark?useSSL=false&serverTimezone=UTC","project_branches_backup1", connectionProperties);

加載JDBC驅動異常信息

Exceptionin thread "main" java.sql.SQLException: No suitable driveratjava.sql.DriverManager.getDriver(DriverManager.java:315)atorg.apache.spark.sql.execution.datasources.jdbc.JDBCOptions.$anonfun$driverClass$2(JDBCOptions.scala:105)atscala.Option.getOrElse(Option.scala:138)atorg.apache.spark.sql.execution.datasources.jdbc.JDBCOptions.<init>(JDBCOptions.scala:105)atorg.apache.spark.sql.execution.datasources.jdbc.JDBCOptions.<init>(JDBCOptions.scala:35)atorg.apache.spark.sql.execution.datasources.jdbc.JdbcRelationProvider.createRelation(JdbcRelationProvider.scala:32)atorg.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:318)atorg.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:223)atorg.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:211)atorg.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:167)atorg.apache.spark.examples.sql.JavaSQLDataSourceExample.runJdbcDatasetExample(JavaSQLDataSourceExample.java:299)atorg.apache.spark.examples.sql.JavaSQLDataSourceExample.main(JavaSQLDataSourceExample.java:120)atsun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)atsun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)atsun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)atjava.lang.reflect.Method.invoke(Method.java:498)atorg.apache.spark.deploy.JavaMainApplication.start(SparkApplication.scala:52)atorg.apache.spark.deploy.SparkSubmit.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:849)atorg.apache.spark.deploy.SparkSubmit.doRunMain$1(SparkSubmit.scala:167)atorg.apache.spark.deploy.SparkSubmit.submit(SparkSubmit.scala:195)atorg.apache.spark.deploy.SparkSubmit.doSubmit(SparkSubmit.scala:86)atorg.apache.spark.deploy.SparkSubmit$$anon$2.doSubmit(SparkSubmit.scala:924)atorg.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:933)atorg.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

解決

步驟1：修改驅動為com.mysql.cj.jdbc.Driver

.format("jdbc").option("driver","com.mysql.cj.jdbc.Driver").option("url","jdbc:mysql://XXX:3306/XX?useSSL=false&serverTimezone=UTC")

步驟2：在代碼中指明driver對應的jar包(mysql-connector-java-8.0.11.jar)所在路徑：

--driver-class-path./spark-2.4.2-bin-hadoop2.7/jars/mysql-connector-java-8.0.11.jar

由于該jar包本身在$SPARK_HOME/jars目錄下，所以無需（2）中顯示指定也會正常加載），對于不在的，需要通過--driver-class-path指定

集成kafka異常信息

Exceptioninthread"main"org.apache.spark.sql.AnalysisException: Failed to find data source: kafka. Please deploy the applicationasper the deployment sectionof"Structured Streaming + Kafka Integration Guide".;

解決

缺少spark-sql-kafka-0-10_2.11-2.4.0.jar包導致，mvn找到該包，然后下載到$SPARK_HOME/jars目錄下

缺少kafka-client包異常信息

Exceptioninthread"main"java.lang.NoClassDefFoundError: org/apache/kafka/common/serialization/ByteArrayDeserializer

解決

缺client包，執(zhí)行cp -f ./kafka-clients-2.2.0-cp2.jar $SPARK_HOME/jars