Wyodrębnianie `Seq [(String, String, String)]` od iskry DataFrame

Question

Mam iskrę DF z rzędami Seq[(String, String, String)]. Staram się zrobić jakąś flatMap z tym, ale wszystko staram kończy się wyrzuceniem

java.lang.ClassCastException: org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema nie mogą być oddane do scala.Tuple3

mogę się pojedynczy rząd albo wiele rzędów od DF dobrze

df.map{ r => r.getSeq[Feature](1)}.first 

powraca

Seq[(String, String, String)] = WrappedArray([ancient,jj,o], [olympia_greece,nn,location] ..... 

i typ danych RDD wydaje się być poprawny.

org.apache.spark.rdd.RDD[Seq[(String, String, String)]]

Schemat DF jest

root 
|-- article_id: long (nullable = true) 
|-- content_processed: array (nullable = true) 
| |-- element: struct (containsNull = true) 
| | |-- lemma: string (nullable = true) 
| | |-- pos_tag: string (nullable = true) 
| | |-- ne_tag: string (nullable = true) 

Znam ten problem jest związany z iskra sql leczenia wiersze RDD jak org.apache.spark.sql.Row choć idiotycznie powiedzieć, że jest to Seq[(String, String, String)]. Jest powiązane pytanie (link poniżej), ale odpowiedź na to pytanie nie działa dla mnie. Nie jestem też dość znany iskrowi, aby dowiedzieć się, jak przekształcić go w działające rozwiązanie.

Czy wiersze są Row[Seq[(String, String, String)]] lub Row[(String, String, String)] lub Seq[Row[(String, String, String)]] lub coś jeszcze bardziej szalonego.

Próbuję zrobić coś takiego

df.map{ r => r.getSeq[Feature](1)}.map(_(1)._1) 

który wydaje się działać, ale faktycznie nie

df.map{ r => r.getSeq[Feature](1)}.map(_(1)._1).first 

rzuca powyższy błąd. Więc jak mam (na przykład) uzyskać pierwszy element drugiej krotki w każdym rzędzie?

Również WHY iskra została zaprojektowana, aby to zrobić, wydaje się idiotyczne twierdzić, że coś jest jednego rodzaju, podczas gdy w rzeczywistości nie jest i nie może zostać przekonwertowane na zastrzegany typ.

---

Powiązane pytanie: GenericRowWithSchema exception in casting ArrayBuffer to HashSet in DataFrame to RDD from Hive table

Powiązane raport o błędzie: http://search-hadoop.com/m/q3RTt2bvwy19Dxuq1&subj=ClassCastException+when+extracting+and+collecting+DF+array+column+type

Answer 1

Dobrze, że nie twierdzi, że jest to krotka. twierdzi, że jest to struct który mapuje :

import org.apache.spark.sql.Row 

case class Feature(lemma: String, pos_tag: String, ne_tag: String) 
case class Record(id: Long, content_processed: Seq[Feature]) 

val df = Seq(
    Record(1L, Seq(
    Feature("ancient", "jj", "o"), 
    Feature("olympia_greece", "nn", "location") 
)) 
).toDF 

val content = df.select($"content_processed").rdd.map(_.getSeq[Row](0)) 

Musisz znaleźć dokładne zasady mapowania w Spark SQL programming guide.

Od nie jest dokładnie ładna struktura prawdopodobnie będziesz chciał mapować go do czegoś przydatne:

content.map(_.map { 
    case Row(lemma: String, pos_tag: String, ne_tag: String) => 
    (lemma, pos_tag, ne_tag) 
}) 

czyli

content.map(_.map (row => (
    row.getAs[String]("lemma"), 
    row.getAs[String]("pos_tag"), 
    row.getAs[String]("ne_tag") 
))) 

Wreszcie nieco bardziej zwięzłe podejście z Datasets:

df.as[Record].rdd.map(_.content_processed) 

lub

df.select($"content_processed").as[Seq[(String, String, String)]] 

chociaż wydaje się, że w tej chwili jest nieco błędny.

Istnieje ważna różnica w pierwszym podejściu (Row.getAs), a drugim (Dataset.as). Ten pierwszy wyodrębnia obiekty jako Any i stosuje się do nich: Ta ostatnia używa enkoderów do transformacji między typami wewnętrznymi i pożądaną reprezentacją.