如何基于相等性检查在 Spark 中使用内部数组查询嵌套 json

Posted 2023-04-18

【中文标题】如何基于相等性检查在 Spark 中使用内部数组查询嵌套 json

【英文标题】：How to query nested json with internal arrays in Spark on basis of equality check

【发布时间】：2018-09-26 19:41:37

【问题描述】：

我有一个嵌套的 json 结构加载到 spark 的数据框中。它包含几层数组，我试图弄清楚如何通过内部数组中的值来查询这个结构。

示例：考虑以下结构（directors.json 文件）

[
  
    "director": "Steven Spielberg",
    "films": [
      
        "name": "E.T",
        "actors": ["Henry Thomas", "Drew Barrymore"]
      ,
      
        "name": "The Goonies",
        "actors": ["Sean Astin", "Josh Brolin"]
      
    ]
  ,
  
    "director": "Quentin Tarantino",
    "films": [
      
        "name": "Pulp Fiction",
        "actors": ["John Travolta", "Samuel L. Jackson"]
      ,
      
        "name": "Kill Bill: Vol. 1",
        "actors": ["Uma Thurman", "Daryl Hannah"]
      
    ]
  
]

假设我想运行一个查询，该查询将返回特定演员参与的所有电影。如下所示：

val directors = spark.read.json("directors.json")
directors.select($"films.name").where($"films.actors" === "Henry Thomas")

当我在 spark shell 中运行它时，我得到一个异常：

org.apache.spark.sql.AnalysisException: cannot resolve '(`films`.`actors` = 'Henry Thomas')' due to data type mismatch: differing types in '(`films`.`actors` = 'Henry Thomas')' (array<array<string>> and string).;;
'Project [name#128]
+- 'Filter (films#92.actors = Henry Thomas)
   +- AnalysisBarrier
         +- Project [films#92.name AS name#128, films#92]
            +- Relation[director#91,films#92] json

如何正确地进行这样的查询？

有不同的选择吗？如果有，有什么优缺点？

谢谢

编辑

@thebluephantom 这仍然不起作用。得到类似的异常。 我认为这是因为我在另一个数组中有一个数组。这是一个例外：

org.apache.spark.sql.AnalysisException: cannot resolve 'array_contains(`films`.`actors`, 'Henry Thomas')' due to data type mismatch: Arguments must be an array followed by a value of same type as the array members;;
'Filter array_contains(films#7.actors, Henry Thomas)
+- AnalysisBarrier
      +- Project [director#6, films#7]
         +- Relation[director#6,films#7] json

【问题讨论】：

那么，您能否指出答案是否有帮助，如果有，请接受？

我想我希望它更简单，但这很好用。谢谢:)

【参考方案1】：

尝试类似的方法，即必须分解电影数据，这意味着重复的演员组是简单的标准化 - 否则我也无法让它工作 - 也许其他人可以：

使用 SPARK 2.3.1 更完整地使用您的数据如下：

val df = spark.read
   .option("multiLine", true).option("mode", "PERMISSIVE")
   .json("/FileStore/tables/films.txt")

val flattened = df.select($"director", explode($"films").as("films_flat"))
flattened.select ("*").where (array_contains (flattened("films_flat.actors"), "Henry Thomas")).show(false)

返回：

 +----------------+-------------------------------------+
 |director        |films_flat                           |
 +----------------+-------------------------------------+
 |Steven Spielberg|[[Henry Thomas, Drew Barrymore], E.T]|
 +----------------+-------------------------------------+

【讨论】：

仍然失败。在问题正文中添加了新的例外

需要对actors使用explode - 这是有道理的。

嵌套 oi 嵌套，确实可以进行一些改进，也就是说，它仍然是非常热门的东西。成功

【参考方案2】：

问题陈述： 假设我想运行一个查询，该查询将返回特定演员参与的所有电影。

解决方案：

val nested_json = spark.read.option("multiline","true").json("\\directors.json")//path to json file 
import spark.implicits._
val films_directors_df = nested_json.select("director","films.name")
val films_directors_flatten = films_directors_df.withColumn("name",explode($"name")).as("films")
films_directors_flatten.show(false)
val mys = nested_json.select("films")
val films__flatten = mys.withColumn("films",explode($"films"))
val actors_name_df = films__flatten.select("films.actors","films.name")
//actors_name_df.show(false) //contains both actors across film
val actors_name_flatten = actors_name_df.withColumn("actors",explode($"actors")).as("actors")
actors_name_flatten.show(false)
val final_df = films_directors_flatten
  .join(actors_name_flatten,$"films.name"===$"actors.name")
  .select($"films.director" as "directors",$"films.name" as "films",$"actors.actors" as "actors")
final_df.show(false)


final_df is the dataframe where you can query for all films that a specific actor has participated in. This solves your above question.
sample query : final_df.filter($"actors"==="Henry Thomas").show(false)

Cheers!!