在Spark中讀取多個文件並在追加之前處理它

我的問題是我有多個相同結構的txt文件，我想將它導入spark。然後創建一個標識符列，彙總數據並最終堆疊它們。在Spark中讀取多個文件並在追加之前處理它

E.g.該文件的一個如下所示：

Date  A B C 
2/21/2014 A1 11 2 
2/22/2014 A1 11 5 
2/23/2014 A1 21 3 
2/24/2014 A1 13 5 
2/25/2014 A1 23 4 
2/26/2014 A1 28 4 
2/27/2014 A1 32 2 
2/28/2014 B1 45 4 
3/1/2014 B1 39 4 
3/2/2014 B1 29 4 
3/3/2014 B1 49 5 
3/4/2014 B1 18 4 
3/5/2014 B1 30 3 
3/6/2014 B1 50 5

閱讀本文件後，我想增加一列提的文件名和更新的數據是這樣的：

Date A B C File 
2/21/2014 A1 22 2 File1 
2/22/2014 A1 36 2 File1 
2/23/2014 A1 17 4 File1 
2/24/2014 A1 30 2 File1 
2/25/2014 A1 11 2 File1 
2/26/2014 A1 32 2 File1 
2/27/2014 A1 19 5 File1 
2/28/2014 B1 22 3 File1 
3/1/2014 B1 12 5 File1 
3/2/2014 B1 50 3 File1 
3/3/2014 B1 42 4 File1 
3/4/2014 B1 37 4 File1 
3/5/2014 B1 31 5 File1 
3/6/2014 B1 20 3 File1

然後彙總數據：

File A B C 
File1 A1 167 19 
File1 B1 214 27

相似，將創建和彙總另一個數據集。最後堆疊在一起。在2案卷中的數據集是這樣的：

File A B C 
File1 A1 167 19 
File1 B1 214 27 
File2 Z10 167 19 
File2 X20 214 27

我可以單獨將它轉換成數據幀導入數據，對它們進行處理，最後將它們堆疊。但我無法以自動化的方式做到這一點。誰能幫幫我嗎。

非常感謝！

來源

2016-05-24 Beta

您是否嘗試過任何一種邏輯？ – WoodChopper

@WoodChopper：謝謝你的評論。對於自動化，沒有。但我讀取數據，轉換成sql數據框，添加文件名的新列。如果你想，我可以把我迄今爲止使用的代碼放在問題本身中。 – Beta

如果你的單個文件裝入內存，你可以使用wholeTextFiles，

rdd = sc.wholeTextFiles("/directorypath/*") 

def appender(x): 
    i = x[0] 
    j = x[1].split("\n") 
    k = [x.split() for x in j] 
    l = [x.append(i) for x in k] 
    return k 

frdd = rdd.flatMap(appender) 

df = frdd.toDF("Date","A","B","C","FileName")

wholeTextFiles回報（文件名，文件內容）的元組從那裏你可以追加文件名。

df.groupBy("FileName","A").count() ##sum()

來源

2016-05-25 12:41:46 WoodChopper

非常感謝您的回答。這真的很有幫助！ – Beta

在Spark中讀取多個文件並在追加之前處理它

回答

相關問題