Spark处理方法

方法"/>

Spark处理方法

需求描述

        通过读取目录下的类似文件的datapath路径的文件名及文件内容，需要将读取的每一个文件的文件名日期解析出来，并作为读取当前文件内容递归读取当前文件一个df列，列名为“时间”；后面就是读一个文件，解析一下时间，将时间添加到所读文件的dataframe中，相当于给每一个读的文件内容添加了当前读取文件解析出来的时间列，以此递归循环。读取每一个文件后都要加上一列此文件解析出来的时间，形式如下

  Reading file: hdfs://master.test:8020/data/测试文件/_L_20231025095119.txtReading file: hdfs://master.test:8020/data/测试文件/_L_20231026095119.txtReading file: hdfs://master.test:8020/data/测试文件/_L_20231027095119.txt

内容形式如下

  文件名：L_20231025095119.txt文件内容：id，imsi,lon,lat......文件名：L_20231026095119.txt文件内容：id，imsi,lon,lat......文件名：L_20231027095119.txt文件内容：id，imsi,lon,lat......我要的文件内容输出形式为id，imsi,lon,lat，20231025095119...id，imsi,lon,lat，20231026095119...id，imsi,lon,lat，20231027095119......

代码片段：

...
...
//打印读取目录下的文件名
println("文件datapath路径")
if (dataPath.isEmpty) {println("No files to read.")
} else {for (path <- dataPath) {println(s"Reading file: $path")}
}
// 解析时间字段
// 提取文件名中的时间部分，格式为："_L_20231111095119.txt" 中的 "20231111095119"
// 由于文件名是按目录读取的，因此要变成字符串形式并以逗号隔开
val filePath = dataPath.mkString(",")
val fileName = filePath.split("/").last.stripSuffix(".csv")
val timeStr = fileName.split("_").last
val year = timeStr.substring(0, 4)
val month = timeStr.substring(4, 6)
val day = timeStr.substring(6, 8)
val hour = timeStr.substring(8, 10)
val minute = timeStr.substring(10, 12)
val second = timeStr.substring(12, 14)
val time = s"$year-$month-$day $hour:$minute:$second"
println(time)
// 输出时间类似：2023-03-01 09:51:19
...
...
val rddAll = if(dataPath.isEmpty){sc.emptyRDD[(LongWritable,Text)]
} else {sc.newAPIHadoopRDD(jobConf,classOf[CombineTextInputFormat],classOf[LongWritable],classOf[Text])}val firstLine = {if(rddAll.take(1).length == 0){null} else {val firstLineTemp = rddAll.first()._2getAfterSplitArray(delimiter,new String(firstLineTemp.getBytes,0,firstLineTemp.getLength,encoding))}
}
...
...
val outDF = sqlc.createDataFrame(outRDD,schema)
//2修改
val outDF1 = outDF.withColumn("时间", lit(time))
outDF1.show()