使用嵌套列连接两个spark Dataframe并更新其中一个列
使用嵌套列连接两个spark Dataframe并更新其中一个列
提问于 2020-01-13 05:11:18
我正在处理一些需求，其中我从CSV文件中获得了一个小表格，如下所示：
root
|-- ACCT_NO: string (nullable = true)
|-- SUBID: integer (nullable = true)
|-- MCODE: string (nullable = true)
|-- NewClosedDate: timestamp (nullable = true我们还有一个非常大的Avro形式的外部配置单元表，它存储在HDFS中，如下所示：
root
-- accountlinks: array (nullable = true)
| | |-- account: struct (nullable = true)
| | | |-- acctno: string (nullable = true)
| | | |-- subid: string (nullable = true)
| | | |-- mcode: string (nullable = true)
| | | |-- openeddate: string (nullable = true)
| | | |-- closeddate: string (nullable = true)现在，需要根据csv文件中的三列查找外部配置单元表：ACCT_NO - SUBID - MCODE。如果匹配，则使用CSV文件中的NewClosedDate更新accountlinks.account.closeddate。
我已经编写了以下代码来分解所需的列并将其与小表连接，但我不太确定如何使用NewClosedDate更新closeddate字段(对于所有帐户持有人，该字段当前为null )，因为closeddate是一个嵌套列，我不能轻松地使用withColumn来填充它。除此之外，模式和列名不能更改，因为这些文件链接到一些外部配置单元表。
val df = spark.sql("select * from db.table where archive='201711'")
val ExtractedColumn = df
.coalesce(150)
.withColumn("ACCT_NO", explode($"accountlinks.account.acctno"))
.withColumn("SUBID", explode($"accountlinks.account.acctsubid"))
.withColumn("MCODE", explode($"C.mcode"))
val ReferenceData = spark.read.format("csv")
.option("header","true")
.option("inferSchema","true")
.load("file.csv")
val FinalData = ExtractedColumn.join(ReferenceData, Seq("ACCT_NO","SUBID","MCODE") , "left")回答 1
Stack Overflow用户
发布于 2020-01-13 10:12:53
您需要做的就是分解accountlinks数组，然后像这样连接两个数据帧：
val explodedDF = df.withColumn("account", explode($"accountlinks"))
val joinCondition = $"ACCT_NO" === $"account.acctno" && $"SUBID" === $"account.subid" && $"MCODE" === $"account.mcode"
val joinDF = explodedDF.join(ReferenceData, joinCondition, "left")现在，您可以像下面这样更新account结构列，并收集列表以获取数组结构：
val FinalData = joinDF.withColumn("account",
struct($"account.acctno", $"account.subid", $"account.mcode",
$"account.openeddate", $"NewClosedDate".alias("closeddate")
)
)
.groupBy().agg(collect_list($"account").alias("accountlinks"))其思想是创建一个新结构，其中包含account中除closedate之外的所有字段，这些字段是从NewCloseDate列获得的。
如果该结构包含许多字段，则可以使用for-comprehension来获取除关闭日期之外的所有字段，以防止键入所有字段。
页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/59710866
复制相关文章
![[1016]DataFrame一列拆成多列以及一行拆成多行](https://ask.qcloudimg.com/http-save/yehe-4908043/1a1ad986ec8928ff4ce5cc152fbc57b1.png)
相似问题
如何在Spark dataframe中使用嵌套列进行连接
如何遍历spark dataframe列并逐个访问其中的值？
如何比较两个Dataframe并更新其中一个Dataframe中的特定列？
更新spark中的dataframe列
使用空值连接两列spark dataframe
添加站长 进交流群
领取专属 10元无门槛券
AI混元助手 在线答疑
关注 腾讯云开发者公众号
洞察 腾讯核心技术
剖析业界实践案例
腾讯云开发者
