如前所述，您在这里有 3 个选项。
在我的示例中，我使用了接下来的 3 个数据集：
+----+----+----+
|col1|col2|col3|
+----+----+----+
|1 |100 |200 |
|2 |300 |400 |
+----+----+----+
+----+----+----+
|col1|col2|col3|
+----+----+----+
|3 |60 |80 |
|4 |12 |100 |
|5 |20 |10 |
+----+----+----+
+----+----+----+
|col1|col2|col3|
+----+----+----+
|7 |20 |40 |
|8 |30 |40 |
+----+----+----+
您首先创建架构（显式定义架构而不是推断它更快）：
import org.apache.spark.sql.types._
val df_schema =
StructType(
List(
StructField("col1", IntegerType, true),
StructField("col2", IntegerType, true),
StructField("col3", IntegerType, true)))
选项 1：
一次加载所有 CSV：
val df1 = spark
.read
.option("header", "false")
.option("delimiter", ",")
.option("inferSchema", "false")
.schema(df_schema)
.csv("file:///C:/data/*.csv")
然后将您的逻辑应用于按文件名分组的整个数据集。
前提条件：你必须想办法将文件名附加到每个文件中
选项 2：
从目录加载 csv 文件。然后遍历文件并为每个 csv 创建一个数据框。在循环内部将您的逻辑应用于每个 csv。最后在循环结束时将结果附加（联合）到第二个数据帧中，该数据帧将存储您的累积结果。
注意：请注意，大量文件可能会导致非常大的 DAG 和随后的巨大执行计划，为了避免这种情况，您可以持久化当前结果或调用 collect。在下面的示例中，我假设将在每个 bufferSize 迭代中执行 persist 或 collect。您可以根据 csv 文件的数量调整甚至删除此逻辑。
这是第二个选项的示例代码：
import java.io.File
import org.apache.spark.sql.Row
import spark.implicits._
val dir = "C:\\data_csv\\"
val csvFiles = new File(dir).listFiles.filter(_.getName.endsWith(".csv"))
val bufferSize = 10
var indx = 0
//create an empty df which will hold the accumulated results
var bigDf = spark.createDataFrame(spark.sparkContext.emptyRDD[Row], df_schema)
csvFiles.foreach{ path =>
var tmp_df = spark
.read
.option("header", "false")
.option("delimiter", ",")
.option("inferSchema", "false")
.schema(df_schema)
.csv(path.getPath)
//execute your custom logic/calculations with tmp_df
if((indx + 1) % bufferSize == 0){
// If buffer size reached then
// 1. call unionDf.persist() or unionDf.collect()
// 2. in the case you use collect() load results into unionDf again
}
bigDf = bigDf.union(tmp_df)
indx = indx + 1
}
bigDf.show(false)
这应该输出：
+----+----+----+
|col1|col2|col3|
+----+----+----+
|1 |100 |200 |
|2 |300 |400 |
|3 |60 |80 |
|4 |12 |100 |
|5 |20 |10 |
|7 |20 |40 |
|8 |30 |40 |
+----+----+----+
选项 3：
最后一个选项是使用内置的spark.sparkContext.wholeTextFiles。
这是将所有 csv 文件加载到 RDD 中的代码：
val data = spark.sparkContext.wholeTextFiles("file:///C:/data_csv/*.csv")
val df = spark.createDataFrame(data)
df.show(false)
还有输出：
+--------------------------+--------------------------+
|_1 |_2 |
+--------------------------+--------------------------+
|file:/C:/data_csv/csv1.csv|1,100,200 |
| |2,300,400 |
|file:/C:/data_csv/csv2.csv|3,60,80 |
| |4,12,100 |
| |5,20,10 |
|file:/C:/data_csv/csv3.csv|7,20,40 |
| |8,30,40 |
+--------------------------+--------------------------+
spark.sparkContext.wholeTextFiles 将返回一个 key/value RDD，其中 key 是文件路径，value 是文件数据。
这需要额外的代码来提取 _2 的内容，即每个 csv 的内容。在我看来，这将包含有关程序性能和可维护性的开销，因此我会避免它。
如果您需要进一步说明，请告诉我