Problem mit Objekt nicht serialisierbar Klasse: org.apache.hadoop.hbase.io.ImmutableBytesWritable Fehler

Question

Ich versuche, eine große Menge an Daten von Spark zu HBase zu laden. Ich verwende saveAsNewAPIHadoopDataset-Methode.

Ich erstelle ImmutableWritable und Put und speichern, die wie folgt erforderlich ist.

dataframe.mapPartitions { rows => 
     { 
     rows.map { eachRow => 
      { 
      val rowKey = Seq(eachRow.getAs[String]("uniqueId"), eachRow.getAs[String]("authTime")).mkString(",") 
      val put = new Put(Bytes.toBytes(rowKey)); 
      val fields = eachRow.schema.fields; 

      for (i <- 0 until fields.length) { 
       put.addColumn(userCF, Bytes.toBytes(fields(i).name), Bytes.toBytes(String.valueOf(eachRow.get(i)))) 
      } 

      (new ImmutableBytesWritable(Bytes.toBytes(rowKey)), put) 
      } 
     } 
     } 
    }.saveAsNewAPIHadoopDataset(job.getConfiguration) 

Meine Daten sind 30 GB wert und es ist in HDFS in 60 Dateien vorhanden.

Wenn ich den gleichen Job mit 10 Dateien gleichzeitig einreichen, ging alles gut.

Aber, wenn ich alles auf einmal einreiche, gibt es diesen Fehler. Der Fehler ist wirklich frustrierend und ich habe alles Mögliche versucht. Aber ich frage mich wirklich, was es geschafft hat, erfolgreich zu laufen, wenn die Daten 5 GB groß sind und was es zu einem Fehler führte, wenn es 30 GB ist.

Hat jemand mit dieser Art von Problemen konfrontiert.?

Answer 1

Das liegt daran, dass ImmutableBytesWritable nicht serialisierbar ist. Wenn es einen Shuffle gibt, versucht Apache Funke, es zu serialisieren, um es an einen anderen Knoten zu senden. Das gleiche würde passieren, wenn Sie versuchen würden, etwas zu nehmen oder es auf Fahrer zu sammeln.

Es gibt eigentlich nur zwei Ansätze.

• Verwenden Sie es nicht, wenn Sie shuffling. Wenn Sie nur jeden Datensatz von der Festplatte in eine Datenbank einfügen müssen, ist ein Shuffling nicht erforderlich. Stellen Sie sicher, dass es ist. Wenn Sie Ihre Daten vorverarbeiten müssen, bevor sie in die Datenbank gelangen, bewahren Sie sie in einem anderen serialisierbaren Format auf und konvertieren Sie sie beim Speichern in das erforderliche Format.
• Verwenden Sie einen anderen Serializer. Apache spark kommt mit Kryo (stelle sicher, dass du spark 2.0.0 verwendest - Kryo wurde dort aktualisiert und es behebt einige böse Nebenläufigkeitsfehler). Um es zu benutzen, müssen Sie es konfigurieren. Es ist nicht schwer, erfordert aber ein wenig Code.