scala - 避免加入 Spark Scala DataFrame
问题描述
我必须根据位于 Azure Blob 存储文件夹中的 JSON 文件进行计算。我正在 Azure HDInsight 上开发 Apache Spark。
此文件夹有一个与跟踪订单相关的编号。如果存在更高的数字,我必须读取此文件夹的 JSON 并丢弃具有较低数字的文件夹。例如,如果我有一个名为 20200501-1 和 20200501-2 的文件夹,我必须读取 20200501-2。
我在 Apache Spark 中找到的解决方案是读取路径,并在数据框中添加一列,如下所示:
val visits = session.read.schema(schema).json(pathData).withColumn("path", input_file_name())
通过这条路,我做了一些转变。但是,这种转换涉及连接和分组,因此,当我在集群中使用大型数据集运行作业时,Spark 作业会花费大量时间。是否有可能进行不同的转变?或改进我的方法。
我的转换使用这样的数据框(添加列后):
val visits = Seq(
("ITEM4449", 33, "https://somefolder@some.net/20200514-1/somename.json"),
("ITEM4450", 16, "https://somefolder@some.net/20200514-1/somename.json"),
("ITEM1111", 88, "https://somefolder@some.net/20200514-2/somename.json"),
("ITEM4453", 64, "https://somefolder@some.net/20200514-1/somename.json"),
("ITEM1111", 12, "https://somefolder@some.net/20200514-1/somename.json")).
toDF("itemId", "visits", "path")
我做了这个转变:
def discardByTrackingCode(rawDataFrame: DataFrame): DataFrame = {
val visitWithColumn = rawDataFrame.
withColumn("tracking_version",
expr("substring(path, 38, 1)"))
val itemVersionDf = visitWithColumn.
withColumn("item_version",
concat(col("ItemId"), lit("_"), col("tracking_version")))
val versionToTakeDf = itemVersionDf.
groupBy(col("ItemId").as("item_id_delete")).
agg(max("item_version").as("item_version"))
val itemReport = itemVersionDf.join(versionToTakeDf, Seq("item_version"))
val finalDf = itemReport.select("ItemId", "Visits", "item_version")
finalDf
}
并得到如下数据框,正确的是:
+--------+------+------------+
|ItemId |Visits|item_version|
+--------+------+------------+
|ITEM4449|33 |ITEM4449_1 |
|ITEM4450|16 |ITEM4450_1 |
|ITEM1111|88 |ITEM1111_2 |
|ITEM4453|64 |ITEM4453_1 |
+--------+------+------------+
有一种最有效的方法可以使此功能起作用吗?除此之外。是否可以(或更可取)使用Hadoop FileSystem类查找文件夹?
解决方案
您可以尝试使用 Window 表达式:
import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions.Window
val window = Window.partitionBy("itemidnumber").orderBy(desc("fileVersion"))
val visits = Seq(
("ITEM4449", 33, "https://somefolder@some.net/20200514-1/somename.json"),
("ITEM4450", 16, "https://somefolder@some.net/20200514-1/somename.json"),
("ITEM1111", 88, "https://somefolder@some.net/20200514-2/somename.json"),
("ITEM4453", 64, "https://somefolder@some.net/20200514-1/somename.json"),
("ITEM1111", 12, "https://somefolder@some.net/20200514-1/somename.json"))
.toDF("itemId", "visits", "path")
.withColumn("itemidnumber", expr("substring(itemId, 5, 4)"))
.withColumn("fileversion", expr("substring(path, 38, 1)"))
.withColumn("tracking_version", expr("concat(itemidnumber, substring(path, 38, 1))"))
.withColumn("row_number", row_number.over(window))
.filter($"row_number" === 1)
display(visits)
输出:
推荐阅读
- javascript - 4xx 状态代码是否应该被视为前端/SSR 应用程序中的错误?
- jacoco - 如何忽略/排除覆盖类中一个特定方法的 Jacoco Coverage?
- firebase - 我应该如何为我的 React Native 应用程序设计社交登录?
- php - 在 php 中从表中将数据插入数据库时遇到问题
- python - 错误“TypeError:需要一个类似字节的对象,而不是'int'”
- python - 如何使用 python 在 Windows 10 中打印当前壁纸路径?
- assembly - 我们如何在 Apple 平台的 AArch64 汇编中在一行上编写多个语句?
- swift - nw_protocol_get_quic_image_block_invoke dlopen libquic 在从 API 调用获取数据时失败
- python - 无法在 ubuntu 中显示 IPython 对象,但在 colab 输出中显示
- node.js - 为什么当我将文件对象从前端发送到后端时它发生了变化?