在创建Dataframe时，我得到了“nullpointer exception”

aij0ehis  于 2021-07-14  发布在  Spark
关注(0)|答案(0)|浏览(364)

我正在从另一个Dataframe行创建Dataframe。这意味着我有另一个Dataframe，在这里被命名为file\u details。该Dataframe包含此格式的行[“srcpath”，“table\u name”，“primary\u key”，“type\u of \u file”，“separator”]。我正在创建一个函数。该函数逐个获取Dataframe行。在那里，我正在检查文件的类型。如果它是csv，使用spark.read选项，我有该csv文件的src\u路径和分隔符。我创建一个Dataframe，并将该Dataframe写入我的sql server.target表名，主键也是我在一个Dataframe行中得到的，在上面我已经写了行结构。
这是我写的代码。

  1. import org.apache.spark.sql.types._
  2. import org.apache.spark.sql.SparkSession
  3. import org.apache.spark.sql.Row
  4. object fixedlength {
  5.   def main(args: Array[String]): Unit = {
  7.     val spark = SparkSession
  8.       .builder()
  9.       .appName("fixedlength")
  10.       .master("local")
  11.       .getOrCreate()
  12.     val file_details=spark.read.option("header", "true").csv("////")
  13.     def fun (ro:Row) : Unit= {         //Row like that ("srcpath","table_name","primary_key","type_of_file","separator")
  14.       val url="jdbc:mysql://*****:3306/mydb1"
  15.       val typ=ro.getString(3)
  16.     if(typ=="csv")
  17.          {
  19.            val sep=ro.getString(4);
  20.            val src_path=ro.getString(0);
  21.            val tgt_table=ro.getString(1);
  22.            val df1=spark.read.option("header", "true").option("delimiter",sep).csv(src_path)
  23.            df1.write.format("jdbc")
  24.              .mode("overwrite")
  25.              .option("url",url)
  26.              .option("driver","com.mysql.jdbc.Driver")
  27.              .option("dbtable",tgt_table)
  28.              .option("user","username")
  29.              .option("password","password")
  30.              .save()
  31.           }
  32.     }
  33.    val ne=file_details.select("*").rdd.map(x=>fun(x)).collect()
  34. }
  35. }

我犯了个错误，

  1. ERROR Executor: Exception in task 0.0 in stage 7.0 (TID 7)
  2. java.lang.NullPointerException
  3.     at fixedlength$.fixedlength$$fun$1(fixedlength.scala:66)
  4.     at fixedlength$$anonfun$10.apply(fixedlength.scala:106)
  5.     at fixedlength$$anonfun$10.apply(fixedlength.scala:106)
  6.     at scala.collection.Iterator$$anon$11.next(Iterator.scala:410)
  7.     at scala.collection.Iterator$class.foreach(Iterator.scala:891)
  8.     at scala.collection.AbstractIterator.foreach(Iterator.scala:1334)
  9.     at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:59)
  10.     at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:104)
  11.     at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:48)
  12.     at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:310)
  13.     at scala.collection.AbstractIterator.to(Iterator.scala:1334)
  14.     at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:302)
  15.     at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1334)
  16.     at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:289)
  17.     at scala.collection.AbstractIterator.toArray(Iterator.scala:1334)
  18.     at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$13.apply(RDD.scala:945)
  19.     at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$13.apply(RDD.scala:945)
  20.     at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2101)
  21.     at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2101)
  22.     at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
  23.     at org.apache.spark.scheduler.Task.run(Task.scala:121)
  24.     at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:402)
  25.     at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
  26.     at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:408)
  27.     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
  28.     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
  29.     at java.lang.Thread.run(Thread.java:748)
  30. 21/04/12 11:11:09 WARN TaskSetManager: Lost task 0.0 in stage 7.0 (TID 7, localhost, executor driver): java.lang.NullPointerException
  31.     at fixedlength$.fixedlength$$fun$1(fixedlength.scala:66)
  32.     at fixedlength$$anonfun$10.apply(fixedlength.scala:106)
  33.     at fixedlength$$anonfun$10.apply(fixedlength.scala:106)
  34.     at scala.collection.Iterator$$anon$11.next(Iterator.scala:410)
  35.     at scala.collection.Iterator$class.foreach(Iterator.scala:891)
  36.     at scala.collection.AbstractIterator.foreach(Iterator.scala:1334)
  37.     at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:59)
  38.     at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:104)
  39.     at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:48)
  40.     at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:310)
  41.     at scala.collection.AbstractIterator.to(Iterator.scala:1334)
  42.     at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:302)
  43.     at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1334)
  44.     at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:289)
  45.     at scala.collection.AbstractIterator.toArray(Iterator.scala:1334)
  46.     at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$13.apply(RDD.scala:945)
  47.     at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$13.apply(RDD.scala:945)
  48.     at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2101)
  49.     at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2101)
  50.     at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
  51.     at org.apache.spark.scheduler.Task.run(Task.scala:121)
  52.     at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:402)
  53.     at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
  54.     at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:408)
  55.     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
  56.     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
  57.     at java.lang.Thread.run(Thread.java:748)
  59. 21/04/12 11:11:09 ERROR TaskSetManager: Task 0 in stage 7.0 failed 1 times; aborting job
  60. 21/04/12 11:11:09 INFO TaskSchedulerImpl: Removed TaskSet 7.0, whose tasks have all completed, from pool 
  61. 21/04/12 11:11:09 INFO TaskSchedulerImpl: Cancelling stage 7
  62. 21/04/12 11:11:09 INFO TaskSchedulerImpl: Killing all running tasks in stage 7: Stage cancelled
  63. 21/04/12 11:11:09 INFO DAGScheduler: ResultStage 7 (collect at fixedlength.scala:106) failed in 0.156 s due to Job aborted due to stage failure: Task 0 in stage 7.0 failed 1 times, most recent failure: Lost task 0.0 in stage 7.0 (TID 7, localhost, executor driver): java.lang.NullPointerException
  64.     at fixedlength$.fixedlength$$fun$1(fixedlength.scala:66)
  65.     at fixedlength$$anonfun$10.apply(fixedlength.scala:106)
  66.     at fixedlength$$anonfun$10.apply(fixedlength.scala:106)
  67.     at scala.collection.Iterator$$anon$11.next(Iterator.scala:410)
  68.     at scala.collection.Iterator$class.foreach(Iterator.scala:891)
  69.     at scala.collection.AbstractIterator.foreach(Iterator.scala:1334)
  70.     at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:59)
  71.     at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:104)
  72.     at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:48)
  73.     at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:310)
  74.     at scala.collection.AbstractIterator.to(Iterator.scala:1334)
  75.     at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:302)
  76.     at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1334)
  77.     at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:289)
  78.     at scala.collection.AbstractIterator.toArray(Iterator.scala:1334)
  79.     at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$13.apply(RDD.scala:945)
  80.     at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$13.apply(RDD.scala:945)
  81.     at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2101)
  82.     at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2101)
  83.     at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
  84.     at org.apache.spark.scheduler.Task.run(Task.scala:121)
  85.     at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:402)
  86.     at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
  87.     at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:408)
  88.     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
  89.     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
  90.     at java.lang.Thread.run(Thread.java:748)
  92. Driver stacktrace:
  93. 21/04/12 11:11:09 INFO DAGScheduler: Job 7 failed: collect at fixedlength.scala:106, took 0.164149 s
  94. Exception in thread "main" org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 7.0 failed 1 times, most recent failure: Lost task 0.0 in stage 7.0 (TID 7, localhost, executor driver): java.lang.NullPointerException
  95.     at fixedlength$.fixedlength$$fun$1(fixedlength.scala:66)
  96.     at fixedlength$$anonfun$10.apply(fixedlength.scala:106)
  97.     at fixedlength$$anonfun$10.apply(fixedlength.scala:106)
  98.     at scala.collection.Iterator$$anon$11.next(Iterator.scala:410)
  99.     at scala.collection.Iterator$class.foreach(Iterator.scala:891)
  100.     at scala.collection.AbstractIterator.foreach(Iterator.scala:1334)
  101.     at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:59)
  102.     at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:104)
  103.     at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:48)
  104.     at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:310)
  105.     at scala.collection.AbstractIterator.to(Iterator.scala:1334)
  106.     at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:302)
  107.     at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1334)
  108.     at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:289)
  109.     at scala.collection.AbstractIterator.toArray(Iterator.scala:1334)
  110.     at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$13.apply(RDD.scala:945)
  111.     at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$13.apply(RDD.scala:945)
  112.     at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2101)
  113.     at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2101)
  114.     at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
  115.     at org.apache.spark.scheduler.Task.run(Task.scala:121)
  116.     at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:402)
  117.     at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
  118.     at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:408)
  119.     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
  120.     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
  121.     at java.lang.Thread.run(Thread.java:748)
  123. Driver stacktrace:
  124.     at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1887)
  125.     at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1875)
  126.     at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1874)
  127.     at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
  128.     at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
  129.     at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1874)
  130.     at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:926)
  131.     at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:926)
  132.     at scala.Option.foreach(Option.scala:257)
  133.     at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:926)
  134.     at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2108)
  135.     at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2057)
  136.     at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2046)
  137.     at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)
  138.     at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:737)
  139.     at org.apache.spark.SparkContext.runJob(SparkContext.scala:2061)
  140.     at org.apache.spark.SparkContext.runJob(SparkContext.scala:2082)
  141.     at org.apache.spark.SparkContext.runJob(SparkContext.scala:2101)
  142.     at org.apache.spark.SparkContext.runJob(SparkContext.scala:2126)
  143.     at org.apache.spark.rdd.RDD$$anonfun$collect$1.apply(RDD.scala:945)
  144.     at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
  145.     at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
  146.     at org.apache.spark.rdd.RDD.withScope(RDD.scala:363)
  147.     at org.apache.spark.rdd.RDD.collect(RDD.scala:944)
  148.     at fixedlength$.main(fixedlength.scala:106)
  149.     at fixedlength.main(fixedlength.scala)
  150. Caused by: java.lang.NullPointerException
  151.     at fixedlength$.fixedlength$$fun$1(fixedlength.scala:66)
  152.     at fixedlength$$anonfun$10.apply(fixedlength.scala:106)
  153.     at fixedlength$$anonfun$10.apply(fixedlength.scala:106)
  154.     at scala.collection.Iterator$$anon$11.next(Iterator.scala:410)
  155.     at scala.collection.Iterator$class.foreach(Iterator.scala:891)
  156.     at scala.collection.AbstractIterator.foreach(Iterator.scala:1334)
  157.     at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:59)
  158.     at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:104)
  159.     at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:48)
  160.     at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:310)
  161.     at scala.collection.AbstractIterator.to(Iterator.scala:1334)
  162.     at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:302)
  163.     at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1334)
  164.     at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:289)
  165.     at scala.collection.AbstractIterator.toArray(Iterator.scala:1334)
  166.     at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$13.apply(RDD.scala:945)
  167.     at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$13.apply(RDD.scala:945)
  168.     at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2101)
  169.     at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2101)
  170.     at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
  171.     at org.apache.spark.scheduler.Task.run(Task.scala:121)
  172.     at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:402)
  173.     at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
  174.     at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:408)
  175.     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
  176.     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
  177.     at java.lang.Thread.run(Thread.java:748)
  178. 21/04/12 11:11:09 INFO SparkContext: Invoking stop() from shutdown hook
  179. 21/04/12 11:11:09 INFO SparkUI: Stopped Spark web UI at http://LTIN236281.cts.com:4040
  180. 21/04/12 11:11:09 INFO MapOutputTrackerMasterEndpoint: MapOutputTrackerMasterEndpoint stopped!
  181. 21/04/12 11:11:10 INFO MemoryStore: MemoryStore cleared
  182. 21/04/12 11:11:10 INFO BlockManager: BlockManager stopped
  183. 21/04/12 11:11:10 INFO BlockManagerMaster: BlockManagerMaster stopped
  184. 21/04/12 11:11:10 INFO OutputCommitCoordinator$OutputCommitCoordinatorEndpoint: OutputCommitCoordinator stopped!
  185. 21/04/12 11:11:10 INFO SparkContext: Successfully stopped SparkContext
  186. 21/04/12 11:11:10 INFO ShutdownHookManager: Shutdown hook called
  187. 21/04/12 11:11:10 INFO ShutdownHookManager: Deleting directory C:\Users\844605\AppData\Local\Temp\spark-cae157ad-b63e-43b4-831e-9cc61094b4db

我在internet上搜索了那个错误“空指针异常”，我在某个地方发现，当工作节点试图访问spark上下文对象时，会抛出这个错误。
有什么不同的方法来实现这个目标吗？
你能请任何人帮我摆脱这个错误吗？
我的案例研究是，我有一个数据框，它包含文件的src\u路径，文件的类型\u，如果它是该文件的csv分隔符，则是sql中的目标表。我将这些行传递到我创建的一个函数中，它获取这些参数的基础是它创建dataframe并将该dataframe推送到sql目标表中。

apache-sparkapache-spark-sqlspark-streaming

暂无答案！

目前还没有任何答案，快来回答吧！

我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com