如何对比Hive与SAS9.2的Schema DataFrame并实现类型校验？

Got it, let's tackle this schema validation problem step by step. You need to compare the SAS and Hive schema DataFrames, validate the type mappings, and output a non-zero value when all checks pass. Here's how to implement this in Scala with Spark:

Step 1: Define Type Mapping Rules #

First, we formalize the expected mappings between SAS types (plus their formats) and Hive types based on your example:

• SAS Num with DATETIME format → Hive TimestampType
• SAS Num with decimal format (like 15.2) → Hive DoubleType
• SAS Num with no format → Hive IntegerType
• SAS Char → Hive StringType

We'll encode these rules using Spark's conditional functions.

Step 2: Standardize Column Names #

Notice that SAS uses uppercase variable names while Hive uses lowercase. Align them first to avoid join issues:

// Convert SAS variable names to lowercase to match Hive's format
val sasSchemaNormalized = metadata.withColumn("Variable", lower(col("Variable")))

Step 3: Join the Two Schema DataFrames #

Join the normalized SAS schema with the Hive schema on the Variable column to bring type information side by side:

val schemaComparison = sasSchemaNormalized.join(df2, Seq("Variable"), "inner")

Step 4: Add Validation Column #

Create a boolean column is_valid to check if each variable's Hive type matches the expected type from SAS:

import org.apache.spark.sql.functions._

val validatedSchema = schemaComparison.withColumn("is_valid",
  when(
    // Check datetime-formatted numeric columns
    (col("Type") === "Num" && col("Format").isNotNull && col("Format").contains("DATETIME")),
    col("type") === "TimestampType"
  ).when(
    // Check decimal-formatted numeric columns
    (col("Type") === "Num" && col("Format").isNotNull && col("Format").contains(".")),
    col("type") === "DoubleType"
  ).when(
    // Check plain numeric columns (no format)
    (col("Type") === "Num" && col("Format").isNull),
    col("type") === "IntegerType"
  ).when(
    // Check character columns
    col("Type") === "Char",
    col("type") === "StringType"
  ).otherwise(false)
)

Step 5: Validate and Output Result #

Count the number of valid mappings. If all variables pass validation (valid count equals total count), output a non-zero value (like the valid count itself). If any fail, print the failing entries and output 0 for clarity:

val totalColumns = validatedSchema.count()
val validColumns = validatedSchema.filter(col("is_valid")).count()

if (validColumns == totalColumns) {
  println(validColumns) // Output non-zero value when all checks pass
} else {
  // Print failing schema entries for debugging
  println("Schema validation failed for the following columns:")
  validatedSchema.filter(!col("is_valid")).select("Variable", "Type", "Format", "type").show()
  println(0)
}

Notes #

• Adjust the format checks (like contains("DATETIME") or contains(".")) if your SAS data uses different format patterns.
• If you need to handle extra columns in either schema, modify the join type from inner to full_outer and add checks for missing entries.

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