Spark SQL写入MySQL时表Schema被覆盖的解决方案咨询

Hey there! Let's work through this issue since you're using Spark 1.6 (a bit legacy, but we can get this right). The core problem here is how Spark 1.6's JDBC writer handles SaveMode.Overwrite, plus some incorrect type conversion syntax in your code.

Why Your Current Approach Isn't Working #

1. Unassigned DataFrame Transformations: Every time you call selectExpr, Spark returns a new DataFrame—but you never reassign it to intoSql. So your original DataFrame with the original schema is still being written, not the transformed one.
2. Wrong CAST Target Types: java.sql.Types.VARCHAR isn't a valid type in Spark SQL's CAST function. Spark uses its own type identifiers (like string for text types, timestamp for datetime types).
3. Overwrite Mode Recreates Table: Spark 1.6's SaveMode.Overwrite drops the existing MySQL table and recreates it using the DataFrame's schema. That's why your table gets converted to TEXT types—Spark maps its StringType to MySQL's TEXT by default when creating tables.

Step-by-Step Solution #

1. Correctly Transform DataFrame Types #

First, fix your type conversions and make sure to reassign the transformed DataFrame:

// Assume originalDataFrame is your initial DataFrame with the original schema
DataFrame intoSql = originalDataFrame.selectExpr(
    "cast(Field0 as string) Field0",  // Maps to MySQL VARCHAR(20)
    "cast(Field1 as string) Field1",
    "Field2",  // Already DoubleType, no conversion needed
    // Convert string timestamp to Spark TimestampType (matches MySQL DATETIME)
    // If your Field3 has a non-standard format, use unix_timestamp to parse it:
    // "cast(unix_timestamp(Field3, 'yyyy-MM-dd HH:mm:ss') as timestamp) Field3"
    "cast(Field3 as timestamp) Field3"
);

• Note: Ensure your Field3 string matches a format Spark can parse as timestamp (default is yyyy-MM-dd HH:mm:ss). If not, use unix_timestamp with your custom format to convert it first.

2. Avoid SaveMode.Overwrite—Truncate Then Append #

Instead of letting Spark recreate the table, manually truncate the existing table first, then append your transformed data. This preserves the original MySQL schema:

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.Statement;
import org.apache.spark.sql.SaveMode;

// Your JDBC configuration
String url = "jdbc:mysql://your-host:3306/your-db";
String tableName = "your-target-table";
Properties props = new Properties();
props.setProperty("user", "your-user");
props.setProperty("password", "your-password");

// Step 1: Truncate the MySQL table to clear existing data
Connection conn = DriverManager.getConnection(url, props);
Statement stmt = conn.createStatement();
stmt.execute("TRUNCATE TABLE " + tableName);
stmt.close();
conn.close();

// Step 2: Append the transformed DataFrame to the existing table
intoSql.write()
       .mode(SaveMode.Append)
       .jdbc(url, tableName, props);

3. Additional Checks #

• Field Name Matching: Ensure your DataFrame's column names exactly match the MySQL table's field names (case sensitivity depends on your MySQL configuration).
• VARCHAR Length: Make sure the strings in Field0 and Field1 don't exceed 20 characters—otherwise MySQL will truncate them or throw an error (depending on your SQL mode).
• Driver Compatibility: Use a MySQL JDBC driver compatible with Spark 1.6 and your MySQL version (e.g., mysql-connector-java 5.1.x for older MySQL versions).

Why This Works #

By truncating the table manually, you avoid Spark dropping and recreating it. Appending the transformed DataFrame uses the existing table's schema, so your VARCHAR(20) and DATETIME types stay intact.

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