我想在从DBFS(Databricks FileSystem)读取的每个文件上运行以下代码。我在文件夹中的所有文件上测试了它,但我想逐个对文件夹中的每个文件进行类似的计算:
// 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")
非常感谢你提出的想法:)
如上所述,这里有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
将返回一个键/值RDD,其中key是文件路径,value是文件数据。
这需要额外的代码来提取_2的内容,这是每个csv的内容。在我看来,这将包括关于程序的性能和可维护性的开销,因此我会避免它。
如果您需要进一步澄清,请与我们联系