Работа с несколькими столбцами RDD в Spark?

Question

Я прочитал несколько примеров Spark, и я не могу понять, как создать RDD с столбцом столбцов и несколькими столбцами значений из файла CSV.

Я немного читал о Spark SQL и не думаю, что это то, что я хочу в этом случае. Я не ищу интерактивный анализ с этими данными, больше обработки пакетного типа.

Меня интересует синтаксис Java или Scala.

Можете ли вы указать мне правильное направление?

Answer 1

Multiple колонка РДД

Там нет такого понятия, на самом деле, но и вы не необходимости один. Вы можете создать RDD объектов с любым типом T. Этот тип должен моделировать запись, поэтому запись с несколькими столбцами может иметь тип Array[String], Seq[AnyRef] или любые лучшие модели ваших данных. В Scala лучший выбор (для безопасности типов и считывания кода) обычно использует класс case, который представляет собой запись.

Например, если ваш CSV выглядит следующим образом:

+---------+-------------------+--------+-------------+ 
| ID  | Name    | Age | Department | 
+---------+-------------------+--------+-------------+ 
| 1  | John Smith  | 29  | Dev   | 
| 2  | William Black  | 31  | Finance  | 
| 3  | Nancy Stevens  | 32  | Dev   | 
+---------+-------------------+--------+-------------+ 

Вы могли бы, например:

case class Record(id: Long, name: String, age: Int, department: String) 

val input: RDD[String] = sparkContext.textFile("./file") 
val parsed: RDD[Record] = input.map(/* split string and create new Record */) 

Теперь вы можете удобно выполнять преобразования на этом РДУ, например, если вы хотите превратить его в PairRDD с идентификатором, как ключ, просто вызовите keyBy:

val keyed: RDD[(Int, Record)] = parsed.keyBy(_.id) 

Сказанное - хотя вас больше интересует «пакетная обработка», а не анализ - это может быть достигнуто более легко (и, возможно, лучше, в зависимости от того, что вы делаете с этим RDD), используя API DataFrames - это имеет хорошие возможности для безопасного чтения CSV (например, spark-csv) и для обработки данных в виде столбцов без необходимости создавать классы классов, соответствующие каждому типу записи.