如何使用 Spark 从 DBFS 目录加载和处理多个 csv 文件

Question

我想运行我从 DBFS（Databricks 文件系统）读取的每个文件上的以下代码。我对文件夹中的所有文件进行了测试，但我想对文件夹中的每个文件进行类似的计算，一个一个地计算：

// a-e are calculated fields
val df2=Seq(("total",a,b,c,d,e)).toDF("file","total","count1","count2","count3","count4")

//schema is now an empty dataframe
val final1 = schema.union(df2)

这可能吗？我想从 dbfs 读取它也应该与我现在所做的不同：

val df1 = spark
      .read
      .format("csv")
      .option("header", "true")
      .option("delimiter",",")
      .option("inferSchema", "true")
      .load("dbfs:/Reports/*.csv")
      .select("lot of ids")

提前非常感谢您的想法:)

Answer 1

如前所述，这里有 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和随后的巨大执行计划，为了避免这种情况，您可以保留当前结果或致电对方付费。在下面的示例中，我假设每次 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将return一个key/valueRDD，其中key是文件路径，value是文件数据。

这需要额外的代码来提取 _2 的内容，即每个 csv 的内容。在我看来，这将包含有关程序性能和可维护性的开销，因此我会避免它。

如果您需要进一步说明，请告诉我

Answer 2

我正在添加@Alexandros Biratsis 提供的答案。可以使用下面的第一种方法，将文件名连接为同一数据框中的单独列，该数据框中包含来自多个文件的所有数据。

val df1 = spark  
      .read  
      .option("header", "false")  
      .option("delimiter", ",")  
      .option("inferSchema", "false")  
      .schema(df_schema)  
      .csv("file:///C:/data/*.csv")  
      .withColumn("FileName",input_file_name())

这里input_file_name()是一个函数，将文件名添加到DataFrame中的每一行。这是 spark 中的内置函数。
要使用此功能，您需要导入以下命名空间。
导入 org.apache.spark.sql.functions._

可以在 https://spark.apache.org/docs/1.6.0/api/java/org/apache/spark/sql/functions.html

找到函数的文档

我建议不要使用@Alexandros Biratsis 建议的第二种方法，即合并和保留临时数据帧，因为它适用于少量文件，但随着文件数量的增加，它变得太慢，有时它超时并且驱动程序意外关闭。

感谢 Alexandros 的回答，因为这给了我解决问题的方法。

如何使用 Spark 从 DBFS 目录加载和处理多个 csv 文件

How to load and process multiple csv files from a DBFS directory with Spark

csv

scala

dataframe

apache-spark

databricks