Spark estrarre valori da una fila

Question

Ho il seguente dataframe

val transactions_with_counts = sqlContext.sql(
    """SELECT user_id AS user_id, category_id AS category_id, 
    COUNT(category_id) FROM transactions GROUP BY user_id, category_id""") 

Sto cercando di convertire i file di oggetti Voto ma dato x (0) restituisce un array questo fallisce

val ratings = transactions_with_counts 
    .map(x => Rating(x(0).toInt, x(1).toInt, x(2).toInt)) 

error: value toInt is not a member of Any

Answer 1

Iniziamo con alcuni dati dummy:

val transactions = Seq((1, 2), (1, 4), (2, 3)).toDF("user_id", "category_id") 

val transactions_with_counts = transactions 
    .groupBy($"user_id", $"category_id") 
    .count 

transactions_with_counts.printSchema 

// root 
// |-- user_id: integer (nullable = false) 
// |-- category_id: integer (nullable = false) 
// |-- count: long (nullable = false) 

Ci sono alcuni modi per accedere valori e mantenere tipi attesi:

1. pattern matching

   import org.apache.spark.sql.Row 
   
   transactions_with_counts.map{ 
       case Row(user_id: Int, category_id: Int, rating: Long) => 
       Rating(user_id, category_id, rating) 
   } 
   

2. tipizzati get* metodi come getInt, getLong:

   transactions_with_counts.map(
       r => Rating(r.getInt(0), r.getInt(1), r.getLong(2)) 
   ) 
   

3. getAs metodo che può utilizzare sia nomi che indici:

   transactions_with_counts.map(r => Rating(
       r.getAs[Int]("user_id"), r.getAs[Int]("category_id"), r.getAs[Long](2) 
   )) 
   

   Può essere utilizzato per estrarre correttamente i tipi definiti dall'utente, incluso mllib.linalg.Vector. Ovviamente l'accesso per nome richiede uno schema.

4. conversione a tipizzazione statica Dataset (Spark 1.6 +/2.0+):

   transactions_with_counts.as[(Int, Int, Long)] 
   

Answer 2

Utilizzando i set di dati è possibile definire Valutazioni come segue:

case class Rating(user_id: Int, category_id:Int, count:Long) 

della classe di rating qui ha un nome di colonna 'conto' invece di 'valutazione' come suggerito da zero323. Così la variabile rating viene assegnato come segue:

val transactions_with_counts = transactions.groupBy($"user_id", $"category_id").count 

val rating = transactions_with_counts.as[Rating] 

In questo modo non verrà eseguito in errori di runtime in Spark perché il tuo Valutazione nome della colonna classe è identico al nome della colonna 'count' generato dalla Spark sulla pista -tempo.