Q: Конвертировать RDD в Dataframe (java.lang.String не является допустимым внешним типом для схемы int)

Question

Я пытаюсь преобразовать RDD в Dataframe без использования case-класса.CSV-файл выглядит следующим образом:

3,193080,De Gea <br>
0,158023,L. Messi <br>
4,192985,K. De Bruyne <br>
1,20801,Cristiano Ronaldo <br>
2,190871,Neymar Jr <br>


val players = sc.textFile("/Projects/Downloads/players.csv").map(line => line.split(',')).map(r => Row(r(1),r(2),r(3)))
# players: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[230] at map at <console>:34

val schema = StructType(List(StructField("id",IntegerType),StructField("age",IntegerType),StructField("name",StringType)))
# schema: org.apache.spark.sql.types.StructType = StructType(StructField(id,IntegerType,true), StructField(age,IntegerType,true), StructField(name,StringType,true))

val playersDF = spark.createDataFrame(players,schema)
# playersDF: org.apache.spark.sql.DataFrame = [id: int, age: int ... 1 more field]

Все идет хорошо, пока я не попытаюсь, например, сделать PlayersDF.show

java.lang.RuntimeException: Error while encoding: java.lang.RuntimeException: java.lang.String is not a valid external type for schema of int 

Что я могу сделать?

Answer 1

Я думаю, что лучший вариант - предоставить схему и прочитать файл csv, используя существующие средства .

import org.apache.spark.sql.types._

val playerSchema = StructType(Array(
    StructField("id", IntegerType, true),
    StructField("age", IntegerType, true),
    StructField("name", StringType, true)
))

val players = spark
    .sqlContext
    .read
    .format("csv")
    .option("delimiter", ",")
    .schema(playerSchema)
    .load("/mypath/players.csv")

Вот результат:

scala> players.show
+---+------+-----------------+
| id|   age|             name|
+---+------+-----------------+
|  3|193080|           De Gea|
|  0|158023|         L. Messi|
|  4|192985|     K. De Bruyne|
|  1| 20801|Cristiano Ronaldo|
|  2|190871|        Neymar Jr|
+---+------+-----------------+

scala> players.printSchema()
root
 |-- id: integer (nullable = true)
 |-- age: integer (nullable = true)
 |-- name: string (nullable = true)

scala>

Answer 2

У вас есть две проблемы:

1) Ваш индекс отключен;Скала основана на 0.Row(r(1),r(2),r(3)) должно быть Row(r(0),r(1),r(2)).

2) line.split возвращает Array[String], в то время как ваша схема указывает, что первое и второе поля должны быть целыми числами.Вам нужно привести их к целым числам перед созданием фрейма данных.

По сути, это то, как вы должны создать players:

val players = rdd.map(line => line.split(","))
                 .map(r => Row(r(0).toInt, r(1).toInt, r(2)))