Как разделить строки на разные столбцы в Spark DataFrame/DataSet?

Question

Предположим, у меня есть набор данных, как:

Name | Subject | Y1 | Y2 
A | math | 1998| 2000 
B |   | 1996| 1999 
    | science | 2004| 2005 

Я хочу разделить строки этого набора данных таким образом, что столбец У2 будут устранены, как:

Name | Subject | Y1 
A | math | 1998 
A | math | 1999 
A | math | 2000 
B |   | 1996 
B |   | 1997 
B |   | 1998 
B |   | 1999 
    | science | 2004 
    | science | 2005 

Может кто-то предложить что-то здесь? Надеюсь, я сделал свой запрос понятным. Заранее спасибо.

Answer 1

Вы можете легко исправить выбор, чтобы получить то, что вы хотите, в кадре данных или даже в RDD.

Dataset<Row> sqlDF = spark.sql("SELECT Name,Subject,Y1 FROM tableName"); 

, если вы начинаете с уже exesting кадра данных, говорят пользователи, вы можете использовать что-то вроде этого:

resultDF = usersDF.select("Name","Subject","Y1"); 

Answer 2

Я тестировал этот код в pyspark и он работает, как ожидалось:

data= sc.parallelize([["A","math",1998,2000],["B","",1996,1999],["","science",2004,2005]] 

data.map(lambda reg: ((reg[0],reg[1]),(range(reg[2],reg[3]+1)))) 
    .flatMapValues(lambda reg: reg).collect() 

Более подробно вам необходимо преобразовать входные данные в пару RDD в форме (ключ, значение), где ключ состоит из первых двух полей, так как результат будет сплющен, сохраняя ключ в целости с flatMapValues. Значения, подлежащие отображению, строятся как range от Y1 до Y2. Все это делается в первом map.

flatMapValues будет возвращать каждое из значений в range, связанных с его key.

Результат выглядит следующим образом:

[(('A', 'math'), 1998), 
(('A', 'math'), 1999), 
(('A', 'math'), 2000), 
(('B', ''), 1996), 
(('B', ''), 1997), 
(('B', ''), 1998), 
(('B', ''), 1999), 
(('', 'science'), 2004), 
(('', 'science'), 2005)] 

Answer 3

Я думаю, вам нужно всего лишь создать udf создать диапазон. Затем вы можете использовать взрываются, чтобы создать необходимые строки:

val createRange = udf { (yearFrom: Int, yearTo: Int) => 
    (yearFrom to yearTo).toList 
} 

df.select($"Name", $"Subject", functions.explode(createRange($"Y1", $"Y2"))).show() 

EDIT: версия питона этого кода будет что-то вроде:

from pyspark.sql import Row 
from pyspark.sql.functions import udf, explode 
from pyspark.sql.types import IntegerType 

createRange=udf(lambda (yearFrom, yearTo): list(range(yearFrom, yearTo)), IntegerType()) 

df.select($"Name", $"Subject", explode(createRange($"Y1", $"Y2"))).show() 

Answer 4

Вот способ, в котором вы можете осуществить это:

val resultantDF= df.rdd.flatMap{row => 
    val rangeInitial = row.getInt(2) 
    val rangeEnd = row.getInt(3) 
    val array = rangeInitial to rangeEnd 
    (List.fill(array.size)(row.getString(0)),List.fill(array.size)(row.getString(1)),array).zipped.toList 
    }.toDF("Name","Subject","Y1") 

resultantDF.show()