Конвертировать CSV в JSON для сопряжения RDD в Scala Spark

Question

У меня есть данные CSV. Сначала я хочу преобразовать его в Json, а затем преобразовать в Pair RDD.

Мне удалось сделать обе вещи, но я не уверен, насколько эффективно это сделать, а также ключ не в ожидаемом формате.

    val df = //some how read the csv data
    val dataset = df.toJSON //This gives the expected json.
    val pairRDD = dataset.rdd.map(record => (JSON.parseFull(record).get.asInstanceOf[Map[String, String]].get("hashKey"), record))

Предположим, моя схема

    root
     |-- hashKey: string (nullable = true)
     |-- sortKey: string (nullable = true)
     |-- score: number (nullable = true)
     |-- payload: string (nullable = true)


    In json
    {
    "hashKey" : "h1",
    "sortKey" : "s1",
    "score" : 1.0,
    "payload" : "data"
    }
    {
    "hashKey" : "h2",
    "sortKey" : "s2",
    "score" : 1.0,
    "payload" : "data"
    }

    EXPECTED result should be
    [1, {"hashKey" : "1", "sortKey" : "2", "score" : 1.0, "payload" : "data"} ]
    [2, {"hashKey" : "h2", "sortKey" : "s2", "score" : 1.0, "payload" : "data"}]


    ACTUAL result I am getting
    [**Some(1)**, {"hashKey" : "1", "sortKey" : "2", "score" : 1.0, "payload" : "data"} ]
    [**Some(2)**, {"hashKey" : "h2", "sortKey" : "s2", "score" : 1.0, "payload" : "data"}]

Можно это исправить?

Answer 1

это из-за get("hashKey").Измените его на getOrElse("hashKey","{defaultKey}") - когда ваш ключ по умолчанию может быть "" или что-то константное, которое вы объявили ранее.

Обновление для более безопасного скала (вместо использования instance of)

Лучше изменить синтаксический анализ JSON следующим образом:

dataset.rdd.map(record => JSON.parseFull(record).map{
    case json: Map[String, String] => (json.getOrElse("hashKey",""), record)
    case _ => ("", "")
}.filter{ case (key, record) => key != "" && record != "") }