Scastie-Rendering-Compilerfehler als "Wert countByValue ist kein Mitglied von org.apache.spark.sql.Dataset [String]"
Hallo, ich versuche, ein Bewertungshistogramm mit dem Scastie-Programm zu finden ... hier ist die Implementierung
sbet Einstellungen in Scastie
scalacOptions ++= Seq(
"-deprecation",
"-encoding", "UTF-8",
"-feature",
"-unchecked"
)
libraryDependencies ++= Seq(
"org.apache.spark" %% "spark-core" % "2.4.3",
"org.apache.spark" %% "spark-sql" % "2.4.3"
)
aktueller Code in Scastie
import org.apache.spark.sql.SparkSession
import org.apache.spark._
import org.apache.spark.SparkContext._
import org.apache.spark.sql.SparkSession
import org.apache.log4j._
object TestApp extends App {
lazy implicit val spark =
SparkSession.builder().master("local").appName("spark_test").getOrCreate()
import spark.implicits._ // Required to call the .toDF function later
val html = scala.io.Source.fromURL("http://files.grouplens.org/datasets/movielens/ml-
100k/u.data").mkString // Get all rows as one string
val seqOfRecords = html.split("\n") // Split based on the newline characters
.filter(_ != "") // Filter out any empty lines
.toSeq // Convert to Seq so we can convert to DF later
.map(row => row.split("\t"))
.map { case Array(f1,f2,f3,f4) => (f1,f2,f3,f4) }
val df = seqOfRecords.toDF("col1", "col2", "col3", "col4")
val ratings = df.map(x => x.toString().split("\t")(2))
// Count up how many times each value (rating) occurs
val results = ratings.countByValue()
// Sort the resulting map of (rating, count) tuples
val sortedResults = results.toSeq.sortBy(_._1)
// Print each result on its own line.
sortedResults.foreach(println)
spark.close()
}
Fehler beim Einsteigen in Scastie
value countByValue ist kein Mitglied von org.apache.spark.sql.Dataset [String]
kann jemand beim deubggen helfen
=========================================== Überarbeiteter Code mit unterschiedlichen Fehlern in Scastie jetzt
java.lang.ExceptionInInitializerError
at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:152)
at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:127)
at org.apache.spark.sql.execution.TakeOrderedAndProjectExec.executeCollect(limit.scala:136)
at org.apache.spark.sql.Dataset.org$apache$spark$sql$Dataset$$collectFromPlan(Dataset.scala:3383) at org.apache.spark.sql.Dataset$$anonfun$head$1.apply(Dataset.scala:2544)
at org.apache.spark.sql.Dataset$$anonfun$head$1.apply(Dataset.scala:2544) at org.apache.spark.sql.Dataset$$anonfun$53.apply(Dataset.scala:3364) at org.apache.spark.sql.execution.SQLExecution$$anonfun$withNewExecutionId$1.apply(SQLExecution.scala:78)
at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:125) at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:73)
at org.apache.spark.sql.Dataset.withAction(Dataset.scala:3363)
at org.apache.spark.sql.Dataset.head(Dataset.scala:2544)
at org.apache.spark.sql.Dataset.take(Dataset.scala:2758)
at org.apache.spark.sql.Dataset.getRows(Dataset.scala:254)
at org.apache.spark.sql.Dataset.showString(Dataset.scala:291)
at org.apache.spark.sql.Dataset.show(Dataset.scala:745)
at org.apache.spark.sql.Dataset.show(Dataset.scala:704)
at org.apache.spark.sql.Dataset.show(Dataset.scala:713)
at TestApp$.delayedEndpoint$TestApp$1(main.scala:22) at TestApp$delayedInit$body.apply(main.scala:4) at scala.Function0$class.apply$mcV$sp(Function0.scala:34)
at scala.runtime.AbstractFunction0.apply$mcV$sp(AbstractFunction0.scala:12)
at scala.App$$anonfun$main$1.apply(App.scala:76) at scala.App$$anonfun$main$1.apply(App.scala:76)
at scala.collection.immutable.List.foreach(List.scala:392)
at scala.collection.generic.TraversableForwarder$class.foreach(TraversableForwarder.scala:35) at scala.App$class.main(App.scala:76)
at TestApp$.main(main.scala:4) at TestApp.main(main.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 sbt.Run.invokeMain(Run.scala:115) at sbt.Run.execute$1(Run.scala:79)
at sbt.Run.$anonfun$runWithLoader$4(Run.scala:92) at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23) at sbt.util.InterfaceUtil$$anon$1.get(InterfaceUtil.scala:10) at sbt.TrapExit$App.run(TrapExit.scala:257)
at java.lang.Thread.run(Thread.java:748)
Caused by: com.fasterxml.jackson.databind.JsonMappingException: Incompatible Jackson version: 2.9.8
at com.fasterxml.jackson.module.scala.JacksonModule$class.setupModule(JacksonModule.scala:64) at com.fasterxml.jackson.module.scala.DefaultScalaModule.setupModule(DefaultScalaModule.scala:19) at com.fasterxml.jackson.databind.ObjectMapper.registerModule(ObjectMapper.java:751) at org.apache.spark.rdd.RDDOperationScope$.<init>(RDDOperationScope.scala:82)
at org.apache.spark.rdd.RDDOperationScope$.<clinit>(RDDOperationScope.scala)
... 40 more
Hier ist der Code in Scastie aktualisiert
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.col
object TestApp extends App {
lazy implicit val spark = SparkSession.builder().master("local").appName("spark_test").getOrCreate()
import spark.implicits._ // Required to call the .toDF function later
val html = scala.io.Source.fromURL("http://files.grouplens.org/datasets/movielens/ml-100k/u.data").mkString // Get all rows as one string
val seqOfRecords = html.split("\n") // Split based on the newline characters
.filter(_ != "") // Filter out any empty lines
.toSeq // Convert to Seq so we can convert to DF later
.map(row => row.split("\t")) // Split each line on tab character to make an Array of 4 String each
.map { case Array(f1,f2,f3,f4) => (f1,f2,f3,f4) } // Convert that Array[String] into Array[(String, String, String, String)]
val df = seqOfRecords.toDF("col1", "col2", "col3", "col4") // Give whatever column names you want
df.select("col3").groupBy("col3").count.sort(col("count").desc).show()
spark.close() // don't forget to close(), otherwise scastie won't let you create another session so soon.
}
Antworten
Erster Teil Ihrer Frage: Das Hauptproblem in Ihrem Code ist also der Versuch, nach Registerkarten aufzuteilen \t. Ihre Datensätze enthalten keine Registerkarten, wie ich in meinem Kommentar erklärt habe.
Die Sache ist, wenn Sie durch die Karte
df, Sie jedes zugreifenorg.apache.spark.sql.RowObjekt, zBdf.firstist[196,242,3,881250949]. Sie können dies in a umwandelnString, aber es gibt kein\t(Tabulatorzeichen), durch das geteilt werden kann. Daher wird einfach eins zurückgegeben,Stringwie es in einemArray[String]mit nur einem Element der Fall ist. Wenn Sie also auf das zweite Element zugreifen, wird ein zurückgegebenjava.lang.ArrayIndexOutOfBoundsException.
Hier ist eine Demonstration:
// We get the first row and brute force convert it toString()
df.head.toString
//res21: String = [196,242,3,881250949] <- See? No tab anywhere
df.head.toString.split("\t")
//res22: Array[String] = Array([196,242,3,881250949]) <- Returns the string as is in an Array
res22(0)
//res24: String = [196,242,3,881250949] <- First Element
res22(1)
//java.lang.ArrayIndexOutOfBoundsException: 1 <- No second (or third) element found, hence the "out of bounds" exception.
// ... 55 elided
Ich habe aus Ihrem Kommentar verstanden, dass Sie versuchen, die dritte Spalte zu erhalten. Das Schöne an der Verwendung von a DataFrameist, dass Sie einfach selectdie gewünschte Spalte nach Namen auswählen können . Sie können es dann groupBy(dies gibt ein RelationalGroupedDataset zurück ) und die countMethode zum Aggregieren verwenden.
import org.apache.spark.sql.functions.col
df.select("col3").groupBy("col3").count.sort(col("count").desc).show()
//+----+-----+
//|col3|count|
//+----+-----+
//| 4|34174|
//| 3|27145|
//| 5|21201|
//| 2|11370|
//| 1| 6110|
//+----+-----+
Zweiter Teil Ihrer Frage: Es scheint, dass Scastie eine neuere Version von com.fasterxml.jackson.core:jackson-databindSpark 2.4.3 lädt. Während Scastie anscheinend die Version verwendet 2.9.6, verwendet Spark 2.4.3 eine ältere Version : 2.6.7.
Die einzige Möglichkeit, es zum Laufen zu bringen, war die Verwendung einer neueren Version von Spark und Scala. Spark 3.0.1 verwendet 2.10.0.
Unter Build-Einstellungen:
- Set
Scala Versionzu2.12.10. - Festlegen der Abhängigkeiten der zusätzlichen Sbt-Konfigurationsbibliothek:
libraryDependencies ++= Seq(
"org.apache.spark" %% "spark-core" % "3.0.1",
"org.apache.spark" %% "spark-sql" % "3.0.1"
)
Es funktioniert nicht so gut, der Browser hängt und manchmal läuft es ab. Ich denke, Scastie ist noch nicht für diese Version optimiert.
Bearbeiten: Eigentlich funktioniert es jetzt viel besser, nachdem ich die Protokollierung zum Schweigen gebracht habe !
ABER immer noch ... Sie sollten Spark wirklich auf Ihrem lokalen Computer installieren .
Wenn Sie die ratingsVariable erreichen, arbeiten Sie mit einer Spark-Struktur, die als Dataset bezeichnet wird. Sie können an der Dokumentation aussehen zu beschreiben , was es kann und was nicht hier . Es gibt keine Methode countByValue, weshalb Sie den angezeigten Fehler erhalten.
Alles, was Sie haben, macht Sinn, bis Sie zu dieser Zeile kommen:
val ratings = df.map(x => x.toString().split("\t")(2))
Dies wird derzeit einen Fehler erzeugen.
Wenn Sie zur dfVariablen zurückkehren, haben Sie eine Tabelle, die ungefähr so aussieht:
+----+----+----+---------+
|col1|col2|col3| col4|
+----+----+----+---------+
| 196| 242| 3|881250949|
| 186| 302| 3|891717742|
| 22| 377| 1|878887116|
| 244| 51| 2|880606923|
| 166| 346| 1|886397596|
+----+----+----+---------+
Sie können den Befehl ausführen df.show(), um ein Beispiel für das zu sehen, was sich im Dataset befindet. Von dort aus möchten Sie eine Operation, die ein bisschen aussieht groupBy. Schauen Sie sich einige Beispiele dafür an, um zu sehen, wohin Sie als nächstes gehen müssen.