如何根据传入参数从键值对RDD中提取指定值？

嘿，我来帮你搞定这个需求！你已经有了以Merchant_Name为键的RDD，值是包含Display_Name和Store_ID_name的JSON字符串，要提取对应字段的话，核心是先把JSON字符串解析成可操作的结构，然后再根据键查询提取。下面分步骤给你实现方案：

步骤1：修复并解析JSON字符串 #

注意你的JSON字符串格式不标准（键没有双引号），得先修复格式再解析。我们可以用Scala自带的JSON解析库，这里先展示原生库的实现：

import scala.util.parsing.json.JSON

// 先把原RDD的JSON值解析成可访问的Map结构
val parsedRDD = mapdf.mapValues(jsonStr => {
  // 修复JSON格式：给键加上双引号，符合标准JSON规范
  val fixedJson = jsonStr.replaceAll("(Display_Name|Store_ID_name)", "\"$1\"")
  // 解析JSON字符串，返回Option[Map[String, Any]]，避免解析失败的情况
  JSON.parseFull(fixedJson).asInstanceOf[Option[Map[String, Any]]]
})

步骤2：实现获取Display_Name的函数 #

这个函数接收Merchant_Name字符串，返回对应的Display_Name（包装在Option里，处理键不存在的情况）：

def getDisplayName(merchantName: String): Option[String] = {
  // lookup方法根据键获取所有对应值的Seq，取第一个值（如果存在）
  parsedRDD.lookup(merchantName).headOption.flatMap { jsonMapOpt =>
    // 从解析后的Map中提取Display_Name字段，转换为String类型
    jsonMapOpt.flatMap(map => map.get("Display_Name").asInstanceOf[Option[String]])
  }
}

步骤3：实现获取Store_ID_name的函数 #

类似地，这个函数返回对应的Store_ID_name，专门处理字段为null的场景：

def getStoreId(merchantName: String): Option[String] = {
  parsedRDD.lookup(merchantName).headOption.flatMap { jsonMapOpt =>
    jsonMapOpt.flatMap(map => map.get("Store_ID_name") match {
      case null | None => Some(null) // 处理字段为null的情况
      case Some(value) => Some(value.asInstanceOf[String])
    })
  }
}

优化方案：转换为本地Map提升查询性能 #

如果你的RDD数据量不大，建议把RDD转换成本地Map，这样每次查询不用重复触发RDD计算，性能更优：

// 将RDD转换为本地Map，仅执行一次action操作
val merchantLocalMap: Map[String, Option[Map[String, Any]]] = parsedRDD.collectAsMap().toMap

// 基于本地Map重构函数
def getDisplayName(merchantName: String): Option[String] = {
  merchantLocalMap.get(merchantName).flatMap(_.flatMap(_.get("Display_Name").asInstanceOf[Option[String]]))
}

def getStoreId(merchantName: String): Option[String] = {
  merchantLocalMap.get(merchantName).flatMap(_.flatMap(map => map.get("Store_ID_name") match {
    case null | None => Some(null)
    case Some(value) => Some(value.asInstanceOf[String])
  }))
}

测试效果 #

调用函数测试你的示例数据：

println(getDisplayName("Amul")) // 输出: Some(Amul)
println(getStoreId("Amul"))      // 输出: Some(null)

println(getDisplayName("Acme ")) // 输出: Some(Acme Fresh Market)
println(getStoreId("Acme "))     // 输出: Some(Acme Markets)

可选：用Case Class提升类型安全性 #

如果想让代码更规范、类型更安全，可以定义一个Case Class来存储解析后的信息：

case class MerchantInfo(displayName: String, storeId: String)

// 重新解析RDD为键值对，值为Option[MerchantInfo]
val parsedRDDWithCaseClass = mapdf.mapValues(jsonStr => {
  val fixedJson = jsonStr.replaceAll("(Display_Name|Store_ID_name)", "\"$1\"")
    .replace("Display_Name", "displayName")
    .replace("Store_ID_name", "storeId") // 匹配Case Class的字段名
  JSON.parseFull(fixedJson).asInstanceOf[Option[Map[String, Any]]].map { map =>
    MerchantInfo(
      map.get("displayName").asInstanceOf[String],
      map.get("storeId").asInstanceOf[String]
    )
  }
})

// 基于Case Class的查询函数
def getDisplayName(merchantName: String): Option[String] = {
  parsedRDDWithCaseClass.lookup(merchantName).headOption.flatMap(_.map(_.displayName))
}

def getStoreId(merchantName: String): Option[String] = {
  parsedRDDWithCaseClass.lookup(merchantName).headOption.flatMap(_.map(_.storeId))
}

这样代码可读性和类型安全性都会更好~

内容的提问来源于stack exchange，提问作者Arjun