Как запросить вложенный JSON с внутренними массивами в Spark на основе проверки равенства

Question

У меня есть вложенная структура json, загруженная в фрейм данных в искре.Он содержит несколько слоев массивов, и я пытаюсь выяснить, как запросить эту структуру по значениям во внутренних массивах.

Пример: рассмотрим следующую структуру (файл 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")

Когда я запускаю это вИскривление оболочки Я получаю исключение:

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

Answer 1

Попробуйте что-то похожее на это, в результате чего данные фильма должны быть разбиты, что означает, что повторяющаяся группа актеров просто нормализована - иначе я тоже не могу заставить ее работать - возможно, кто-то еще может:

Более полно, используя SPARK2.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]|
 +----------------+-------------------------------------+