爲什麼分區鍵列從DataFrame丟失

Question

我有一個作業，加載一個DataFrame對象，然後使用DataFrame partitionBy方法將數據保存爲parquet格式。然後我發佈創建的路徑，以便後續作業可以使用輸出。輸出中的路徑如下所示：

/ptest/_SUCCESS 
/ptest/id=0 
/ptest/id=0/part-00000-942fb247-1fe4-4147-a41a-bc688f932862.snappy.parquet 
/ptest/id=0/part-00001-942fb247-1fe4-4147-a41a-bc688f932862.snappy.parquet 
/ptest/id=0/part-00002-942fb247-1fe4-4147-a41a-bc688f932862.snappy.parquet 
/ptest/id=1 
/ptest/id=1/part-00003-942fb247-1fe4-4147-a41a-bc688f932862.snappy.parquet 
/ptest/id=1/part-00004-942fb247-1fe4-4147-a41a-bc688f932862.snappy.parquet 
/ptest/id=1/part-00005-942fb247-1fe4-4147-a41a-bc688f932862.snappy.parquet 
/ptest/id=3 
/ptest/id=3/part-00006-942fb247-1fe4-4147-a41a-bc688f932862.snappy.parquet 
/ptest/id=3/part-00007-942fb247-1fe4-4147-a41a-bc688f932862.snappy.parquet 

當我接收到新數據時，它會附加到數據集中。路徑已發佈，因此依賴於數據的作業只能處理新數據。

下面的代碼的一個簡單的例子：

>>> rdd = sc.parallelize([(0,1,"A"), (0,1,"B"), (0,2,"C"), (1,2,"D"), (1,10,"E"), (1,20,"F"), (3,18,"G"), (3,18,"H"), (3,18,"I")]) 
>>> df = sqlContext.createDataFrame(rdd, ["id", "score","letter"]) 
>>> df.show() 
+---+-----+------+ 
| id|score|letter| 
+---+-----+------+ 
| 0| 1|  A| 
| 0| 1|  B| 
| 0| 2|  C| 
| 1| 2|  D| 
| 1| 10|  E| 
| 1| 20|  F| 
| 3| 18|  G| 
| 3| 18|  H| 
| 3| 18|  I| 
+---+-----+------+ 
>>> df.write.partitionBy("id").format("parquet").save("hdfs://localhost:9000/ptest") 

問題是，當另一個程序試圖使用已發佈的路徑來讀取文件：

>>> df2 = spark.read.format("parquet").schema(df2.schema).load("hdfs://localhost:9000/ptest/id=0/") 
>>> df2.show() 
+-----+------+ 
|score|letter| 
+-----+------+ 
| 1|  A| 
| 1|  B| 
| 2|  C| 
+-----+------+ 

正如你所看到的分區鍵從加載的數據集中丟失。如果我要發佈作業可以使用的模式，我可以使用模式加載文件。該文件加載和分區鍵存在，但值爲空：

>>> df2 = spark.read.format("parquet").schema(df.schema).load("hdfs://localhost:9000/ptest/id=0/") 
>>> df2.show() 
+----+-----+------+ 
| id|score|letter| 
+----+-----+------+ 
|null| 1|  A| 
|null| 1|  B| 
|null| 2|  C| 
+----+-----+------+ 

有沒有一種方法，以確保分區密鑰存儲瓦特/拼花數據？我不想要求其他進程分析獲取密鑰的路徑。

Answer 1

如果這樣你應該提供basePathoption：

(spark.read 
    .format("parquet") 
    .option("basePath", "hdfs://localhost:9000/ptest/") 
    .load("hdfs://localhost:9000/ptest/id=0/")) 

它指向您的數據的根目錄。

隨着basePathDataFrameReader將意識到分區並相應地調整模式。

Answer 2

如果其他應用程序加載特定分區，它看起來像load("hdfs://localhost:9000/ptest/id=0/")路徑，該應用程序可以調整代碼來替換空與分區列值

part = 0 # partition to load 
df2 =spark.read.format("parquet")\ 
       .schema(df.schema)\ 
       .load("ptest/id="+str(part)).fillna(part,["id"]) 

這樣，輸出將是 -

+---+-----+------+ 
| id|score|letter| 
+---+-----+------+ 
| 0| 1|  A| 
| 0| 1|  B| 
| 0| 2|  C| 
+---+-----+------+