Spark中載入本地(或者hdfs)檔案以及 spark使用SparkContext例項的textFile讀取多個資料夾(巢狀)下的多個數據檔案

在正常呼叫過程中，難免需要對多個資料夾下的多個檔案進行讀取，然而之前只是明確了spark具備讀取多個檔案的能力。
針對多個資料夾下的多個檔案，以前的做法是先進行資料夾的遍歷，然後再進行各個資料夾目錄的讀取，其實不必那麼麻煩，因為spark原生就支援這樣的能力。
原理也非常簡單，就是textFile功能。編寫這樣的程式碼，讀取上次輸出的多個結果，由於RDD儲存結果都是儲存為一個資料夾。而多個相關聯RDD的結果就是多個資料夾。

（1）通過如下程式碼：

  //## read all files(files in different directorys)
          val alldata = sc.textFile("data/Flag/*/part-*")
          println(alldata.count())   
 

（2）textFile檔案路徑傳入格式

預設是從hdfs讀取檔案，也可以指定sc.textFile("路徑").在路徑前面加上hdfs://表示從hdfs檔案系統上讀

本地檔案讀取 sc.textFile("路徑").在路徑前面加上file:// 表示從本地檔案系統讀，如file:///home/user/spark/README.md

具體格式如下：

sc = SparkContext(conf=conf)
 '''# hdfs目錄格式如下'''
input_data_path = "hdfs://localhost:9002/input/2017-11*"
 '''# 本地檔案目錄'''
input_data_path="file:///Users/a6/Downloads/input_local/2017-09*"
print input_data_path
result = sc.textFile(input_data_path)
 

1）hdfs://localhost:9002/input/下目錄結構如下圖：

localhost:userid_hbsid_map_new a6$ hadoop dfs -ls hdfs://localhost:9002/input/
DEPRECATED: Use of this script to execute hdfs command is deprecated.
Instead use the hdfs command for it.

17/11/08 16:30:15 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Found 3 items
-rw-r--r--   1 a6 supergroup        100 2017-11-07 16:41 hdfs://localhost:9002/input/2017-11-01.txt
-rw-r--r--   1 a6 supergroup        100 2017-11-07 16:41 hdfs://localhost:9002/input/2017-11-10.txt
drwxr-xr-x   - a6 supergroup          0 2017-11-08 15:17 hdfs://localhost:9002/input/test_input
localhost:userid_hbsid_map_new a6$ hadoop dfs -ls hdfs://localhost:9002/input/*
DEPRECATED: Use of this script to execute hdfs command is deprecated.
Instead use the hdfs command for it.

17/11/08 16:30:27 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
-rw-r--r--   1 a6 supergroup        100 2017-11-07 16:41 hdfs://localhost:9002/input/2017-11-01.txt
-rw-r--r--   1 a6 supergroup        100 2017-11-07 16:41 hdfs://localhost:9002/input/2017-11-10.txt
Found 1 items
-rw-r--r--   1 a6 supergroup        100 2017-11-08 15:17 hdfs://localhost:9002/input/test_input/2017-11-20.txt
localhost:userid_hbsid_map_new a6$
 

2）下面是在python編寫的spark程式中測試的結果，判斷是否達到預期。 

'''# hdfs目錄'''
（1）input_data_path = "hdfs://localhost:9002/input/2017-11-01.txt"  #一個檔案路徑，這時候只裝載指定的檔案
（2）input_data_path = "hdfs://localhost:9002/input"  #這個報錯，沒有測試通過  —— 一個目錄路徑，這時候只裝載指定目錄下面的所有檔案（不包括子目錄下面的檔案）
（3）input_data_path = "hdfs://localhost:9002/input/*"   #通過萬用字元的形式載入多個檔案或者載入多個目錄下面的所有檔案

py4j.protocol.Py4JJavaError: An error occurred while calling o18.partitions.
: java.io.IOException: Not a file: hdfs://localhost:9002/input/test_input
	at org.apache.hadoop.mapred.FileInputFormat.getSplits(FileInputFormat.java:320)

/user/hdfs/input/dt=20130728/hr=00/
/user/hdfs/input/dt=20130728/hr=01/
...
/user/hdfs/input/dt=20130728/hr=23/
具體的資料都在hr等於某個時間的目錄下面，現在我們要分析20130728這一天的資料，我們就必須把這個目錄下面的所有hr=*的子目錄下面的資料全部裝載進RDD，於是我們可以這樣寫：sc.textFile("hdfs://n1:8020/user/hdfs/input/dt=20130728/hr=*/"),注意到hr=*,是一個模糊匹配的方式。

4）利用Transformations 操作函式union來合併多個檔案的輸入

'''合併輸入多個數據檔案，並集操作，將源資料集與union中的輸入資料集取並集，
    預設保留重複元素'''
    input_data_path_1 = "hdfs://localhost:9002/input/2017-11-01.txt"
    result1 = sc.textFile(input_data_path_1)
    input_data_path_2= "hdfs://localhost:9002/input/2017-11-10.txt"   
    result2 = sc.textFile(input_data_path_2)
    result = result1.union(result2)

union(otherDataset) 

並集操作，將源資料集與union中的輸入資料集取並集，預設保留重複元素（如果不保留重複元素，可以利用distinct操作去除，下邊介紹distinct時會介紹）。

>>> data1 = sc.parallelize(range(10))
>>> data2 = sc.parallelize(range(6,15))
>>> data1.union(data2).collect()
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 6, 7, 8, 9, 10, 11, 12, 13, 14]

‍