将数据帧的顺序保存到 HDFS

Posted 2023-04-15

【中文标题】将数据帧的顺序保存到 HDFS

【英文标题】：Saving Order of a Dataframe to HDFS

【发布时间】：2018-05-11 22:49:16

【问题描述】：

输入数据：

key,date,value    
10,20180701,a10    
11,20180702,a11    
12,20180702,a12    
13,20180702,a13    
14,20180702,a14    
15,20180702,a15    
16,20180702,a16
17,20180702,a17    
18,20180702,a18    
19,20180702,a19    
1 ,20180701,a1     
2 ,20180701,a2     
3 ,20180701,a3     
4 ,20180701,a4     
5 ,20180701,a5     
6 ,20180701,a6 
7 ,20180701,a7 
8 ,20180701,a8 
9 ,20180701,a9 

代码

val rawData=sc.textFile(.....).
val datadf:DataFrame=rawData.toDF

将数据读入带有key,data,value列的DF后

datadf.coalesce(1).orderBy(desc("key")).drop(col("key")).write.mode("overwrite").partitionBy("date").text("hdfs://path/")

我正在尝试按列键对列进行排序，并在保存到 hdfs 之前删除同一列（每天放入一个文件中）。 我无法保留输出文件中的顺序。 如果我不使用合并，订单将被保留，但会生成多个文件。

输出：

/20180701/part-xxxxxxx.txt

        a1
        a9
        a6
        a4
        a5
        a3
        a7
        a8
        a2
        a10
/20180702/part-xxxxxxx.txt  

        a18
        a12
        a13
        a19
        a15
        a16
        a17
        a11
        a14

预期的 OP：

/20180701/part-xxxxxxx.txt

        a1
        a2
        a3
        a4
        a5
        a6
        a7
        a8
        a9
        a10
/20180702/part-xxxxxxx.txt      

        a11
        a12
        a13
        a14
        a15
        a16
        a17
        a18
        a19

【问题讨论】：

“key”字段的类型是什么？

键域类型为字符串

【参考方案1】：

以下代码应该可以帮助您入门（这是使用 Spark 2.1）：-

import org.apache.spark.sql.types.StructType
val schema = new StructType().add($"key".int).add($"date".string).add($"value".string)
val df = spark.read.schema(schema).option("header","true").csv("source.txt")
df.coalesce(1).orderBy("key").drop("key").write.mode("overwrite").partitionBy("date").csv("hdfs://path/")