如何通过加入 Spark 创建嵌套列？

Question

我想在两个 Spark DataFrames (Scala) 上执行“连接”，但不是类似 SQL 的连接，我想将第二个 DataFrame 中的“连接”行作为单个嵌套列插入首先。这样做的原因最终是为了写回具有嵌套结构的 JSON。我知道答案可能已经在 Stackoverflow 上，但是一些搜索还没有找到我的答案。

表 1

root
 |-- Insdc: string (nullable = true)
 |-- LastMetaUpdate: string (nullable = true)
 |-- LastUpdate: string (nullable = true)
 |-- Published: string (nullable = true)
 |-- Received: string (nullable = true)
 |-- ReplacedBy: string (nullable = true)
 |-- Status: string (nullable = true)
 |-- Type: string (nullable = true)
 |-- accession: string (nullable = true)
 |-- alias: string (nullable = true)
 |-- attributes: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- tag: string (nullable = true)
 |    |    |-- value: string (nullable = true)
 |-- center_name: string (nullable = true)
 |-- design_description: string (nullable = true)
 |-- geo_accession: string (nullable = true)
 |-- instrument_model: string (nullable = true)
 |-- library_construction_protocol: string (nullable = true)
 |-- library_name: string (nullable = true)
 |-- library_selection: string (nullable = true)
 |-- library_source: string (nullable = true)
 |-- library_strategy: string (nullable = true)
 |-- paired: boolean (nullable = true)
 |-- platform: string (nullable = true)
 |-- read_spec: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- base_coord: long (nullable = true)
 |    |    |-- read_class: string (nullable = true)
 |    |    |-- read_index: long (nullable = true)
 |    |    |-- read_type: string (nullable = true)
 |-- sample_accession: string (nullable = true)
 |-- spot_length: long (nullable = true)
 |-- study_accession: string (nullable = true)
 |-- tags: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- title: string (nullable = true)

表 2

root
 |-- BioProject: string (nullable = true)
 |-- Insdc: string (nullable = true)
 |-- LastMetaUpdate: string (nullable = true)
 |-- LastUpdate: string (nullable = true)
 |-- Published: string (nullable = true)
 |-- Received: string (nullable = true)
 |-- ReplacedBy: string (nullable = true)
 |-- Status: string (nullable = true)
 |-- Type: string (nullable = true)
 |-- abstract: string (nullable = true)
 |-- accession: string (nullable = true)
 |-- alias: string (nullable = true)
 |-- attributes: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- tag: string (nullable = true)
 |    |    |-- value: string (nullable = true)
 |-- dbGaP: string (nullable = true)
 |-- description: string (nullable = true)
 |-- external_id: struct (nullable = true)
 |    |-- id: string (nullable = true)
 |    |-- namespace: string (nullable = true)
 |-- submitter_id: struct (nullable = true)
 |    |-- id: string (nullable = true)
 |    |-- namespace: string (nullable = true)
 |-- tags: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- title: string (nullable = true)

加入table1.study_accession 和table2.accession。结果如下。请注意名为 study 的新列，其中包含表 2 中行的等效记录。

root
 |-- Insdc: string (nullable = true)
 |-- LastMetaUpdate: string (nullable = true)
 |-- LastUpdate: string (nullable = true)
 |-- Published: string (nullable = true)
 |-- Received: string (nullable = true)
 |-- ReplacedBy: string (nullable = true)
 |-- Status: string (nullable = true)
 |-- Type: string (nullable = true)
 |-- accession: string (nullable = true)
 |-- alias: string (nullable = true)
 |-- attributes: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- tag: string (nullable = true)
 |    |    |-- value: string (nullable = true)
 |-- center_name: string (nullable = true)
 |-- design_description: string (nullable = true)
 |-- geo_accession: string (nullable = true)
 |-- instrument_model: string (nullable = true)
 |-- library_construction_protocol: string (nullable = true)
 |-- library_name: string (nullable = true)
 |-- library_selection: string (nullable = true)
 |-- library_source: string (nullable = true)
 |-- library_strategy: string (nullable = true)
 |-- paired: boolean (nullable = true)
 |-- platform: string (nullable = true)
 |-- read_spec: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- base_coord: long (nullable = true)
 |    |    |-- read_class: string (nullable = true)
 |    |    |-- read_index: long (nullable = true)
 |    |    |-- read_type: string (nullable = true)
 |-- sample_accession: string (nullable = true)
 |-- spot_length: long (nullable = true)
 |-- study_accession: string (nullable = true)
 |-- tags: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- title: string (nullable = true)
 |-- accession: string (nullable = true)
 |-- study: struct (nullable = true)
 |    |-- BioProject: string (nullable = true)
 |    |-- Insdc: string (nullable = true)
 |    |-- LastMetaUpdate: string (nullable = true)
 |    |-- LastUpdate: string (nullable = true)
 |    |-- Published: string (nullable = true)
 |    |-- Received: string (nullable = true)
 |    |-- ReplacedBy: string (nullable = true)
 |    |-- Status: string (nullable = true)
 |    |-- Type: string (nullable = true)
 |    |-- abstract: string (nullable = true)
 |    |-- accession: string (nullable = true)
 |    |-- alias: string (nullable = true)
 |    |-- attributes: array (nullable = true)
 |    |    |-- element: struct (containsNull = true)
 |    |    |    |-- tag: string (nullable = true)
 |    |    |    |-- value: string (nullable = true)
 |    |-- dbGaP: string (nullable = true)
 |    |-- description: string (nullable = true)
 |    |-- external_id: struct (nullable = true)
 |    |    |-- id: string (nullable = true)
 |    |    |-- namespace: string (nullable = true)
 |    |-- submitter_id: struct (nullable = true)
 |    |    |-- id: string (nullable = true)
 |    |    |-- namespace: string (nullable = true)
 |    |-- tags: array (nullable = true)
 |    |    |-- element: string (containsNull = true)
 |    |-- title: string (nullable = true)

Answer 1

根据我对您的问题的理解，假设您有两个数据框

df1 
root
 |-- col1: string (nullable = true)
 |-- col2: integer (nullable = false)
 |-- col3: double (nullable = false)

和

df2
root
 |-- col1: string (nullable = true)
 |-- col2: string (nullable = true)
 |-- col3: double (nullable = false)

您必须将df2 的所有列组合成struct 列，然后选择要连接的列和struct 列。这里我以col1为加入栏目

import org.apache.spark.sql.functions._
val nestedDF2 = df2.select($"col1", struct(df2.columns.map(col):_*).as("nested_df2"))

最后一步是join（这里默认是inner join）

df1.join(nestedDF2, Seq("col1"))

这应该给你

root
 |-- col1: string (nullable = true)
 |-- col2: integer (nullable = false)
 |-- col3: double (nullable = false)
 |-- nested_df2: struct (nullable = false)
 |    |-- col1: string (nullable = true)
 |    |-- col2: string (nullable = true)
 |    |-- col3: double (nullable = false)

希望回答对你有帮助