- html - 出于某种原因,IE8 对我的 Sass 文件中继承的 html5 CSS 不友好?
- JMeter 在响应断言中使用 span 标签的问题
- html - 在 :hover and :active? 上具有不同效果的 CSS 动画
- html - 相对于居中的 html 内容固定的 CSS 重复背景?
我想对从 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")
提前非常感谢您的想法:)
最佳答案
如前所述,您在这里有 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 的内容。在我看来,这将包含有关程序性能和可维护性的开销,因此我会避免它。
如果您需要进一步说明,请告诉我
关于scala - 如何使用 Spark 从 DBFS 目录加载和处理多个 csv 文件,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/54945058/
我有一个新的 Azure Databricks 实例,正在对其进行一些实验。根据 Databricks 文档,我 activated the DBFS File Browser在管理控制台中。 但是,
我有一个新的 Azure Databricks 实例,正在对其进行一些实验。根据 Databricks 文档,我 activated the DBFS File Browser在管理控制台中。 但是,
我正在使用带有 ADLS 存储层的 Azure Databricks。我怀疑 DBFS 和 Filestore 之间有什么区别?知道吗,Filestore 中可以存储的文件的最大大小是多少?我们可以将
我需要了解一些有关 Databricks DBFS 的信息。 用简单的基本术语来说,它是什么,它的目的是什么以及它允许我做什么? 关于 databricks 的文档,说的是这个效果.. “DBFS
数据库文件系统是一种文件系统,它是数据库而不是层次结构。最初不是一个太复杂的想法,但我想我会问是否有人考虑过他们如何做这样的事情?一个简单的计划可能会遗漏哪些问题?我对实现的第一个猜测是类似于 Lin
我正在尝试使用以下代码使用 python 笔记本将 azure 存储 blob 安装到 azure Databricks 中。 mount_name = '/mnt/testMount' if not
我将文件上传到 DBFS: /FileStore/shared_uploads/name_surname@xxx.xxx/file_name.csv 我试图通过 Pandas 访问它们,但我总是收到此
我想在 Azure Databricks 群集上尝试群集范围的初始化脚本。我正在努力查看哪些命令可用。 基本上,我在 dbfs 上有一个文件,希望在集群启动时将其复制到本地目录 /tmp/config
我想在 Azure Databricks 群集上尝试群集范围的初始化脚本。我正在努力查看哪些命令可用。 基本上,我在 dbfs 上有一个文件,希望在集群启动时将其复制到本地目录 /tmp/config
我正在尝试删除位于 DBFS 中的 Databricks Delta Lake 中的一些记录。我只有路径。没有保存为表格,有没有办法从增量文件中删除记录? 谢谢 最佳答案 更新的答案: Delta L
我将数据上传到Azure Databricks,可以看到DBFS列表中有数据集,但在Databricks所属的资源组中找不到任何数据集。 Azure Databricks DBFS 数据集存储在哪里?
我将数据上传到Azure Databricks,可以看到DBFS列表中有数据集,但在Databricks所属的资源组中找不到任何数据集。 Azure Databricks DBFS 数据集存储在哪里?
我读到here AWS Databricks 上单个文件的存储限制为 5TB,我们可以存储任意数量的文件那么同样的限制也适用于 Azure Databricks 吗?或者,Azure Databric
在调查高数据 block 费用时,我惊奇地发现,其中很多实际上是一个自动创建的存储帐户,具有 GRS 复制到另一个包含大量日志文件(TB 上 TB 数据)的区域例如: dbutils.fs.ls('d
使用python/dbutils,如何在Databricks文件系统(DBFS)中递归显示当前目录和子目录的文件。 最佳答案 关于 dbutils.fs.ls (和 %fs 魔法命令)的令人惊讶的事情
我正在使用 saveAsTextFile() 将 Spark 作业的结果存储在文件夹 dbfs:/FileStore/my_result 中。 我可以使用 Web 浏览器访问不同的“part-xxxx
在azure databricks中,通过简单地添加两个点,我可以得到不同的dbfs目录列表结果。谁能向我解释一下为什么会发生这种情况? 最佳答案 使用 dbutils,您只能使用“dbfs:/”路径
dbutils.fs.mount( source = f"wasbs://{blob.storage_account_container}@{blob.storage_account_name}.
每一个我都很理解。但主要区别是什么? 两者都将数据持久保存在磁盘上吗? 如果我将非分布式系统挂载到数据 block DBFS 会怎样? 最佳答案 我认为在 HDFS 中,数据保存在本地服务器上,但在
我的文件存储在 prod 和 test 环境中的 S3 存储桶中。我的代码将在两个环境的分布式系统中执行。我想从 DBFS 访问文件,因为 DBFS 由 S3 支持,并且它的存储桶安装是指向 S3 位
我是一名优秀的程序员,十分优秀!