Превратите текстовый файл в датафрейм с Scala Spark

Question

У меня есть текстовый файл из S3 (на самом деле несколько файлов .gz), и я написал код ниже

val text = sc.textFile(path)
val df_text = text.map(row => row.split(",")).toDF()

Но результат такой:

+--------------------+
|               value|
+--------------------+
|[id, member_id, l...|
|[1077501, 1296599...|
|[1077430, 1314167...|
|[1077175, 1313524...|
|[1076863, 1277178...|
|[1075358, 1311748...|
|[1075269, 1311441...|
+--------------------+

Я могу ' t читать его как "val df = spark.read.format (" csv "). option (" header "," true "). load (path)", потому что когда я читаю его таким образом, он не может найти заголовок:

+-----------+-----------+-----------+
|1077430    |1356730    |4525526    |...
+-----------+-----------+-----------+
|   41173430|    1356730|    1456430|...
|   10237430|    1356660|    1463750|...
+-----------+-----------+-----------+

Как мне сделать правильный DataFrame?

Answer 1

В версии 2.4.0 с scala 2.12.8.

Это очень просто:

val spark: SparkSession = SparkSession
      .builder
      .master("local[*]")
      .getOrCreate
val sc = spark.sparkContext

val myGZs= sc
      .textFile("s3://route//*.gz")
      .map(parseToObject)
      .filter(obj => obj != null)

val myGZsDF = spark.createDataFrame(myGZs)
myGZsDF.printSchema()

, где parseToObject - это функция типа:

val parseToObject = (row: String) => {
   if (row is header) { //program this
      null
   }
   val split_row = row.split(",")
   Subscription(split_row[0].toLong, split_row[1].toLong ...)
}

Пример Subscription класса дела:

case class Subscription(id: Long, memberId: Long ...)

И map, и filter - это узкие преобразования!

РЕДАКТИРОВАТЬ:

Также видели эту ссылку от @kev о том, как читать несколько файлов GZ и преобразовывать их в DF. Остерегайтесь расширения, оно ДОЛЖНО быть .gz.

Надеюсь, это поможет. Дайте мне знать, если у вас возникнут проблемы. Томас.