Spark结构化流写入时如何将驼峰列名转为下划线小写格式？

Great question! Let's break this down clearly for you:

1. Is there a built-in parameter to map column names directly? #

Unfortunately, no—Spark's Parquet writer (for both batch and Structured Streaming) doesn't include a native configuration option to automatically convert camelCase column names to snake_case. You’ll need to explicitly rename columns before writing the stream.

2. Simplest implementation when column names are unknown #

When you don’t have prior knowledge of the column names, you can create a reusable utility function to handle camelCase-to-snake_case conversion, then apply it to all columns in your Dataset. Here’s a Java implementation that matches your code’s language:

Step 1: Build a camelCase to snake_case converter #

First, add a helper method to transform column names:

private static String camelToSnake(String camelCaseName) {
    // Insert underscore between lowercase and uppercase letters, then convert to full lowercase
    return camelCaseName.replaceAll("([a-z])([A-Z])", "$1_$2").toLowerCase();
}

Step 2: Rename all columns in your Dataset #

Loop through every column in your input Dataset, rename it using the converter, and create a new Dataset with updated column names:

// Start with your original Dataset<Row> ds
Dataset<Row> renamedDs = ds;

// Iterate over each column and apply the name transformation
for (String originalCol : ds.columns()) {
    renamedDs = renamedDs.withColumnRenamed(originalCol, camelToSnake(originalCol));
}

Step 3: Write the renamed Dataset to S3 #

Use the updated Dataset for your stream write operation. Don’t forget to adjust the partitionBy value if your partition column was originally in camelCase (it will now be snake_case):

StreamingQuery query = renamedDs.writeStream()
        .format("parquet")
        .outputMode(OutputMode.Append())
        .option("checkpointLocation", "s3://my-checkpoint-location/")
        .option("path", "s3://my-output-data-location")
        .partitionBy("my_partition_column_name") // Updated to snake_case
        .start();

Key Details: #

• This is a lightweight operation: renaming columns only modifies the Dataset’s metadata, not the underlying data, so it won’t impact stream performance.
• The regex handles standard camelCase patterns (e.g., firstName → first_name, myPartitionColumn → my_partition_column). For edge cases like acronyms (e.g., URLPath → url_path), you can tweak the regex to handle those scenarios.

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