scala spark - saveAsTextFile - 文件名[重复]答案

【问题标题】：scala spark - saveAsTextFile - File name [duplicate]scala spark - saveAsTextFile - 文件名[重复]
【发布时间】：2019-01-21 23:47:58
【问题描述】：

问题陈述：我正在尝试在 Scala 中编写一个 Spark 代码，它将从 HDFS 加载下面提到的两个图块（1. 具有 id 和名称的文件 2. 具有 id 和薪水的文件）并加入相同的，并产生（name.salary）值。并按薪水将数据保存在多个瓦片组中（意味着每个文件将具有相同薪水的员工的名称。文件名也必须包括薪水。

EmployeeName.csv 
E01,Lokesh 
E02,Bhupesh 
E03,Amit 
E04,Ratan 
E05,Dinesh 
E06,Pavan 
07,Tejas 
E08,Sheela 
E09,Kumar
E10,Venkat 

EmployeeSalary.csv 
E01,50000 
E02,50000 
E03,45000 
E04,45000 
E05,50000 
E06,45000 
E07,50000 
E08,10000 
E09,10000 
E10,10000

我尝试了以下，但它没有运行。看起来另一个 RDD 中的 RDD 函数不起作用。我还能怎么解决这个问题？

val employeename = sc.textFile("/user/cloudera/EmployeeName").map(x => (x.split(",")(0),x.split(",")(1)))

val employeesalary = sc.textFile("/user/cloudera/EmployeeSalary").map(s => (s.split(",")(0),s.split(",")(1)))

val join = employeename.join(employeesalary).map({case(id,(name,salary)) => (salary,name)})

val group = join.groupByKey().map({case(key, groupvalues) => {
(key,groupvalues.toList)
}}).sortByKey()`enter code here`

val rdd1 = group.map{case (k,v) => k->sc.parallelize(v)}

rdd1.foreach{case (k,rdd) => rdd.saveAsTextFile("user/cloudera/"+k)}

【问题讨论】：

正确，你不能在另一个 rdd 中有一个 rdd。使用数据框和partitionby 获取不同的文件怎么样？

标签： scala apache-spark

【解决方案1】：

让每个文件（分区）只包含一个员工的信息相对容易，但是，Spark 并没有真正让您控制输出文件名（如Change output filename prefix for DataFrame.write() 此处所述）

请注意，您可以将分区设置为路径的一部分（例如 .../employee=Lokesh/salary=50000/part...txt），但数据不会成为文件的一部分

【讨论】：

【解决方案2】：

*我试过下面的代码sn-ps。

使用 Dataframe 或 Dataset 代替 RDD 会很好地执行操作。

我在 sn-p 下面尝试了这个，它将结果保存在 txt 文件中。默认情况下，它将创建带有分区文件的文件夹。结果可以在文件“part-0000”中查看*

下面是代码sn-p：

val rddInput1 = sc.textFile("Path To Input CSV1").map { x => (x.split(",")(0) -> x.split(",")(1)) }

val rddInput2 = sc.textFile("Path to Input CSV2").map { x => (x.split(",")(0) -> x.split(",")(1)) }

// Join based on EMP_ID
val joinData = rddInput1.join(rddInput2).map(x => x._1 -> x._2._2)

// Reduce By Key to aggregate, Option user can directly use joinData directly
// Actual code which wraps result into RDD then write into txt file

joinData.reduceByKey((k1, k2) => k1 + k2).sortByKey().map(x => x._1 + "__" + x._2).collect().foreach(x => {

val collectionTxt = Array[String](x);

sc.parallelize(collectionTxt)
.repartition(1)
.saveAsTextFile("<Output Dir>" + x)
})

我希望它对你尝试做的事情有所帮助......

【讨论】：