使用时间路径的火花写入操作HDFS

Question

我正在尝试从这个Scala代码写入csv文件。我使用HDFS作为临时目录，然后writer.write在现有子文件夹中创建一个新文件。我收到以下错误消息：

val inputFile = "s3a:/tfsdl-ghd-wb/raidnd/rawdata.csv" //  INPUT path 
val outputFile = "s3a:/tfsdl-ghd-wb/raidnd/Incte_19&20.csv" //  OUTPUT path 
val dateFormat = new SimpleDateFormat("yyyyMMdd")
val fileSystem = getFileSystem(inputFile)
val inputData = readCSVFile(fileSystem, inputFile, skipHeader = true).toSeq

val writer = new PrintWriter(new File(outputFile))
writer.write("Sales,cust,Number,Date,Credit,SKU\n")
filtinp.foreach(x => {
  val (com1, avg1) = com1Average(filtermp, x)
  val (com2, avg2) = com2Average(filtermp, x)
  writer.write(s"${x.Date},${x.cust},${x.Number},${x.Credit}\n")
})
writer.close()

def getFileSystem(path: String): FileSystem = {
val hconf = new Configuration() // initialize new hadoop configuration
new Path(path).getFileSystem(hconf) // get new filesystem to handle data

java.io./tfsdl-ghd-wb/raidnd/Incte_19

如果我选择新建文件或退出文件，也会发生同样的情况，我已经检查了路径是否正确，只想在其中创建一个新文件。

问题是，为了使用基于文件系统的源写入数据，您需要一个临时目录，这是Spark使用的提交机制的一部分，即数据首先写入临时目录，一旦任务完成，就会自动将处理过的文件移动到最终路径。

我是否应该将每个Spark应用程序的临时文件夹路径更改为S3？我认为最好在本地处理（本地文件HDFS），然后将处理后的输出文件上载到S3

此外，我刚刚看到我使用的databricks集群中没有“无火花配置集”，这会干扰问题吗？

Answer 1

如果您能够以DataFrame的形式使用火花/scala读取原始数据，那么您可以对数据帧执行转换以构建最终的数据帧。一旦您有了最终的数据帧，那么需要将其写入csv文件，您可以使用下面一行代码将csv文件保存到s3桶路径或hdfs路径。

df.write.format('csv').option('header','true').mode('overwrite').option('sep',',').save('s3a:/tfsdl-ghd-wb/raidnd/Incte_19&20.csv')