Блог

Мне была предоставлена строка json, которая выглядит следующим образом:

 {
  "dataflows": [
    {
  "name": "test",
  "sources": [
    {
      "name": "person_inputs",
      "path": "/data/input/events/person/*",
      "format": "JSON"
    }
  ],
  "transformations": [
    {
      "name": "validation",
      "type": "validate_fields",
      "params": {
        "input": "person_inputs",
        "validations": [
          {
            "field": "office",
            "validations": [
              "notEmpty"
            ]
          },
          {
            "field": "age",
            "validations": [
              "notNull"
            ]
          }
        ]
      }
    },
    {
      "name": "ok_with_date",
      "type": "add_fields",
      "params": {
        "input": "validation_ok",
        "addFields": [
          {
            "name": "dt",
            "function": "current_timestamp"


          }
        ]
      }
    }
  ],
  "sinks": [
    {
      "input": "ok_with_date",
      "name": "raw-ok",
      "paths": [
        "/data/output/events/person"
      ],
      "format": "JSON",
      "saveMode": "OVERWRITE"
    },
    {
      "input": "validation_ko",
      "name": "raw-ko",
      "paths": [
        "/data/output/discards/person"
      ],
      "format": "JSON",
      "saveMode": "OVERWRITE"
    }
 

И меня попросили использовать его как своего рода рецепт для конвейера ETL, т. Е. Данные должны быть извлечены из «пути», указанного в ключе «источники», преобразования, которые должны быть выполнены, указаны в ключе «преобразования», и, наконец, преобразованные данные должны быть сохранены в одном из двух указанных ключей «приемника».

Я решил преобразовать строку json в карту scala следующим образом:

 val json = Source.fromFile("path/to/json")
//parse
val parsedJson = jsonStrToMap(json.mkString)
implicit val formats = org.json4s.DefaultFormats
val parsedJson = parse(jsonStr).extract[Map[String, Any]]
 

итак, с помощью этого я получаю структуру, подобную этой:

которая представляет собой карту, первым значением которой является список карт. Я могу оценить parsedJson("dataflows") , чтобы получить:

как и ожидалось, это список, но тогда я не могу пройти через такой список, хотя мне это нужно, чтобы добраться до источников, преобразований и приемников. Я попытался использовать индекс списка, чтобы, например, получить его первый элемент , например: parsedJson("dataflows")(0) , но безрезультатно.

Кто-нибудь, пожалуйста, может помочь мне пересечь эту структуру? Любая помощь была бы вам очень признательна.

Овации,

При вычислении parsedJson("dataflows") a Tuple2 возвращается aka a Tuple , который имеет два элемента, к которым осуществляется доступ с ._1 помощью и ._2

Таким dataflows(1)._1 образом,возвращаемое значение является «источниками» и dataflows(1)._2 представляет собой список карт (Список[Карта[K, V]), которые могут быть пройдены, как вы обычно проходите элементы a List , где каждый элемент Map

Давайте разберем это, например:

val dataFlowsZero = ("sources", List(Map(42 -> "foo"), Map(42 -> "bar")))

Первый элемент в Tuple

 scala> dataFlowsZero._1
String = sources
 

Второй элемент в Tuple

 scala> dataFlowsZero._2
List[Map[Int, String]] = List(Map(42 -> foo), Map(42 -> bar))`
 

Сопоставьте ключи в каждом Map List из них с новым List

 scala> dataFlowsZero._2.map(m => m.keys)                                                                                                                                                                    
List[Iterable[Int]] = List(Set(42), Set(42))
 

Сопоставьте значения в каждом Map List из них с новым List

 scala> dataFlowsZero._2.map(m => m.values)
List[Iterable[String]] = List(Iterable(foo), Iterable(bar))
 

Лучшим решением является преобразование JSON в полную структуру данных, которая вам была предоставлена, а не просто Map[String, Any] . Это делает тривиальным выбор нужных вам данных. Например,

 val dataFlows = parse(jsonStr).extract[DataFlows]

case class DataFlows(dataflows: List[DataFlow])
case class DataFlow(name: String, sources: List[Source], transformations: List[Transformation], sinks: List[Sink])
case class Source(name: String, path: String, format: String)
case class Transformation(name: String, `type`: String, params: List[Param])
case class Param(input: String, validations: List[Validation])
case class Validation(field: String, validations: List[String])
case class Sink(input: String, name: String, paths: List[String], format: String, saveMode: String)
 

Идея состоит в том, чтобы заставить обработчик JSON выполнять большую часть работы по созданию типобезопасной версии исходных данных.