如何移除RDD中键的双引号并拆分JSON以适配CEP系统输入？

Hey there! Let's get your RDD data converted into the exact format you need for your CEP system. I'll walk you through two approaches—one that's robust and error-resistant (my top recommendation) and another that uses string manipulation if you want to stick closer to your initial attempt.

Approach 1: JSON Parsing (Recommended) #

Since your input data is valid JSON, using a proper JSON parser is way more reliable than string splitting. It handles edge cases like special characters in values (e.g., a colon inside a string) that would break simple split operations.

Assuming your RDD is of type RDD[String] (each element is a full JSON string), here's how to do this in Scala:

First, set up the Jackson JSON parser (Spark includes this by default, so no extra dependencies needed):

import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.module.scala.DefaultScalaModule

val mapper = new ObjectMapper()
mapper.registerModule(DefaultScalaModule) // Enables Scala-specific type handling

Then apply the map transformation:

val formattedRDD = rdd.map { jsonString =>
  // Parse the JSON string into a Scala Map
  val jsonMap = mapper.readValue[Map[String, Any]](jsonString)
  
  // Extract and format the "t" field
  val timestamp = jsonMap("t").asInstanceOf[String]
  val tSegment = s"""t = "$timestamp""""
  
  // Format all other fields for the Check block (without quotes on keys)
  val checkFields = jsonMap
    .filterKeys(_ != "t") // Exclude the timestamp field
    .map { case (key, value) =>
      // Handle different value types to match your desired format
      value match {
        case str: String => s"""$key = "$str""""
        case num: Number => s"$key = $num"
        case other => s"$key = $other"
      }
    }
    .mkString(", ")
  
  // Combine everything into the final format
  s"$tSegment Check = { $checkFields }"
}

Why this works: #

• It correctly handles string values (adding quotes) and numeric values (keeping them unquoted)
• Resistant to special characters in field values
• Easy to adjust if you add/remove fields later

Approach 2: String Manipulation (Simple Scenarios) #

If you prefer to use string splitting like your initial attempt, we can refine it to avoid errors and match your desired output. This works best if you're certain your field values won't contain ", " or other characters that break splitting.

val formattedRDD = rdd.map { rawLine =>
  // Remove the outer curly braces from the JSON string
  val content = rawLine.stripPrefix("{").stripSuffix("}")
  
  // Split into individual key-value pairs (split on ", " only)
  val allPairs = content.split(", ")
  
  // Separate the "t" field from the rest
  val (tPair, checkPairs) = allPairs.partition(_.startsWith("\"t\""))
  
  // Format the timestamp segment: replace colon with " = "
  val tSegment = tPair.head.replace(":", " = ")
  
  // Format each Check field: remove quotes from keys and replace colon with " = "
  val checkFields = checkPairs.map { pair =>
    // Split only on the first colon to avoid breaking values with colons
    val (quotedKey, value) = pair.split(":", 2)
    val cleanKey = quotedKey.stripPrefix("\"").stripSuffix("\"")
    s"$cleanKey = $value"
  }.mkString(", ")
  
  // Build the final output string
  s"$tSegment Check = { $checkFields }"
}

Test Result #

For your sample input:

{"var":"system-ready","value":0.0,"objectID":"2018","partnumber":2,"t":"2017-08-25 11:27:39.000"}

Both approaches will output exactly what you need:

t = "2017-08-25 11:27:39.000" Check = { var = "system-ready", value = 0.0, objectID = "2018", partnumber = 2 }

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