如何將行拆分爲Spark DataFrame/DataSet中的不同列？

假設我有數據集，如：如何將行拆分爲Spark DataFrame/DataSet中的不同列？

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

我想拆分此數據集這樣的行時，Y2列將被淘汰，如：

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

有人建議可以在這裏的東西？我希望我已經清楚了我的問題。提前致謝。

來源

2016-11-14 neha

@cheseaux你爲什麼要刪除你的答案？這對我來說似乎是有效的。 – eliasah

-1

您可以輕鬆使用火花選擇來獲得您想要的數據框架，甚至是RDD。

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

如果你是從已經exesting數據幀開始，說的用戶，可以使用這樣的事情：

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

來源

2016-11-14 10:21:18 Hasson

查看輸出。主要目的是將行按次數分割爲（Y2-Y1），而不是僅刪除Y2。 – neha

我想根據年份的範圍拆分行，例如，如果數據類似於 ''A |數學| 1998 | 2000' 那麼輸出就像 'A |數學| 1998 A |數學| 1999 A |數學| 2000'' – neha

'import org.apache.spark.sql.functions._ val toRange = udf {（y1：Int，y2：Int）=>（y1 to y2）.toArray} input .withColumn（「years 「，toRange（$」Y1「，$」Y2「）） .select（$」Name「，爆炸（$」years「）爲」Year「）但它顯示NumberFormatException：null，的空值。 – neha

我在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()

更詳細地說，您需要將輸入數據轉換爲形式爲（key，value）的RDD對，其中鍵由前兩個字段組成，因爲結果將變平並保持鍵不變爲flatMapValues。要映射的值構造爲，從Y1到Y2。所有這些都在第一個map完成。

flatMapValues將返回與key相關的range中的每個值。

輸出看起來是這樣的：

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

來源

2016-11-14 11:06:09

我想你只需要創建一個udf創建範圍。然後你可以用爆炸來創造必要的行：

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

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

編輯：此代碼的Python版本會是這樣的：

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()

來源

2016-11-14 11:07:20

我們可以使用python熊貓嗎？我無法理解你的星火代碼。 – neha

這裏是您可以在其中實現這個方式：

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()

來源

2016-11-14 11:12:12

如何將行拆分爲Spark DataFrame/DataSet中的不同列？

回答

相關問題