引起：MatchError: [Ljava.lang.String;@21a536b1 (of class [Ljava.lang.String;) in join 2 Dataframe答案

【问题标题】：Caused by: MatchError: [Ljava.lang.String;@21a536b1 (of class [Ljava.lang.String;) in joining 2 Dataframe引起：MatchError: [Ljava.lang.String;@21a536b1 (of class [Ljava.lang.String;) in join 2 Dataframe
【发布时间】：2021-08-17 07:56:15
【问题描述】：

我正在尝试在 Databricks 环境中使用 Scala 在 Apache Spark 中加入 2 个 Dataframe 在加入这 2 个 Dataframe 时，我遇到了一个错误，我无法弄清楚问题是什么以及如何解决它。非常感谢任何帮助。

第一个输入文件

   %scala
   import org.apache.spark.sql.types._
   import org.apache.spark.sql.functions._

   val rawUserArtistData = sc.textFile("/FileStore/tables/user_artist_data.txt")
   val rawUserArtistDataDF = rawUserArtistData.map(_.split(" ")).map{case Array(a,b,c) => 
   (a.toInt,b.toInt,c.toInt)}.toDF("userid","artist_id","playcount")

   rawUserArtistDataDF.show()

输出

   +-------+---------+---------+
   | userid|artist_id|playcount|
   +-------+---------+---------+
   |1000002|        1|       55|
   |1000002|  1000006|       33|
   |1000002|  1000007|        8|
   |1000002|  1000009|      144|
   |1000002|  1000010|      314|
   |1000002|  1000013|        8|
   |1000002|  1000014|       42|
   |1000002|  1000017|       69|
   |1000002|  1000024|      329|
   |1000002|  1000025|        1|
   +-------+---------+---------+

第二个文件

 %scala
 import org.apache.spark.sql.types._
 import org.apache.spark.sql.functions._

 val rawArtistData = sc.textFile("/FileStore/tables/artist_data.txt")
 val rawArtistDataDF = rawArtistData.map(_.split("\t")).map{case Array(a,b) => 
 (a.toInt,b)}.toDF("artistid","artist_name")

 rawArtistDataDF.show(10, false)

输出

 +--------+---------------------------------+
 |artistid|artist_name                      |
 +--------+---------------------------------+
 |1134999 |06Crazy Life                     |
 |6821360 |Pang Nakarin                     |
 |10113088|Terfel, Bartoli- Mozart: Don     |
 |10151459|The Flaming Sidebur              |
 |6826647 |Bodenstandig 3000                |
 |10186265|Jota Quest e Ivete Sangalo       |
 |6828986 |Toto_XX (1977                    |
 |10236364|U.S Bombs -                      |
 |1135000 |artist formaly know as Mat       |
 |10299728|Kassierer - Musik für beide Ohren|
 +--------+---------------------------------+

加入数据框代码

%scala

val CombinedDF = rawUserArtistDataDF.join(rawArtistDataDF,rawUserArtistDataDF("artist_id") === rawArtistDataDF("artistid"), "leftouter")

CombinedDF.show()

错误

 Job aborted due to stage failure.
 Caused by: MatchError: [Ljava.lang.String;@21a536b1 (of class [Ljava.lang.String;)
 at 
 $line72e2ce7142694dbeb5cc11da58bc59cb37.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.$anonfun$rawArtistDataDF$2(command-1764271964671849:5)
at scala.collection.Iterator$$anon$10.next(Iterator.scala:459)
at scala.collection.Iterator$$anon$10.next(Iterator.scala:459)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage2.processNext(Unknown Source)
at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:754)
at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:458)
at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:155)
at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:73)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:39)
at org.apache.spark.scheduler.Task.doRunTask(Task.scala:148)
at org.apache.spark.scheduler.Task.run(Task.scala:117)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$10(Executor.scala:732)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1643)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:735)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
 Driver stacktrace:
    at org.apache.spark.scheduler.DAGScheduler.failJobAndIndependentStages(DAGScheduler.scala:2766)
    at org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2(DAGScheduler.scala:2713)
    at org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2$adapted(DAGScheduler.scala:2707)
    at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)
    at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)
    at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)
    at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:2707)
    at org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1(DAGScheduler.scala:1256)
    at org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1$adapted(DAGScheduler.scala:1256)
    at scala.Option.foreach(Option.scala:407)
    at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:1256)
    at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2974)
    at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2915)
    at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2903)
    at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)
Caused by: scala.MatchError: [Ljava.lang.String;@21a536b1 (of class [Ljava.lang.String;)
    at $line72e2ce7142694dbeb5cc11da58bc59cb37.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.$anonfun$rawArtistDataDF$2(command-1764271964671849:5)
    at scala.collection.Iterator$$anon$10.next(Iterator.scala:459)
    at scala.collection.Iterator$$anon$10.next(Iterator.scala:459)
    at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage2.processNext(Unknown Source)
    at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
    at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:754)
    at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:458)
    at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:155)
    at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:73)
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:39)
    at org.apache.spark.scheduler.Task.doRunTask(Task.scala:148)
    at org.apache.spark.scheduler.Task.run(Task.scala:117)
    at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$10(Executor.scala:732)
    at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1643)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:735)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)

【问题讨论】：

在你的第二个文件中有一行不能被精确地分割成 2 个字符串。一开始你并没有注意到它，因为你只处理了 10 行。
那么你有什么解决方案必须拆分第二个文件吗？
这取决于。也许您只是想过滤掉分成多于或少于 2 个部分的行。或者，也许您想弄清楚如何更改拆分逻辑。或者，如果有更多，您可能想取前 2 部分，如果更少，则填写默认值。
@Bhavesh 你可以确保你的行都只拆分一次？ val a = List("1\tA","2\tB\tC", "3\t\tC") a.foreach(println) val x = a.map(_.split("\t")) .map(r => { val idx = r(0) val str = r.tail.reduce(_ ++ _) (idx, str) }) print(x)

标签： scala dataframe apache-spark apache-spark-dataset

【解决方案1】：

我的第二个文件有问题，我通过以下方式解决了它

%scala
import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._

val rawArtistData = sc.textFile("/FileStore/tables/artist_data.txt")

val rawArtistDataDF = rawArtistData.flatMap { line =>
  val (id, name) = line.span(_ != '\t')
  if (name.isEmpty) {
    None
  } else {
    try {
      Some((id.toInt, name.trim))
    } catch {
      case _: NumberFormatException => None
    }
  } 
}.toDF("artistid","artist_name")

【讨论】：