Spark SQL中from_unixtime函数输出异常的Java解决方案咨询

问题核心：from_unixtime函数接收的是秒级Unix时间戳，而你的数据中：

• Arrival_Time是毫秒级（13位数字）
• Creation_Time是纳秒级（19位数字）
  直接传入会导致日期计算异常，必须先转换为秒级时间戳再处理。

解决方案 #

1. 处理毫秒级时间戳（Arrival_Time） #

两种实现方式：

• 方式一：将毫秒数除以1000转为秒级，再用from_unixtime
• 方式二：使用to_timestamp函数（Spark 2.2+支持），直接处理毫秒级数值

2. 处理纳秒级时间戳（Creation_Time） #

需要先将纳秒数转换为秒级（除以10^{9）或毫秒级（除以10}6），再转换为日期时间：

• 方式一：纳秒转秒（除以1000000000）后用from_unixtime
• 方式二：纳秒转毫秒（除以1000000）后用to_timestamp

修改后的Java代码 #

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.StructType;
import static org.apache.spark.sql.functions.*;

public class TimeStampConversion {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder()
                .appName("TimeStampConversion")
                .master("local[*]")
                .getOrCreate();

        StructType activitySchema = new StructType()
                .add("Arrival_Time", "BIGINT")
                .add("Creation_Time", "BIGINT")
                .add("Device","string")
                .add("Index", "string")
                .add("Model","string")
                .add("User", "string")
                .add("gt","string")
                .add("x", "DOUBLE")
                .add("y","DOUBLE")
                .add("z","DOUBLE");

        Dataset<Row> jsondf = spark
                .readStream()
                .schema(activitySchema)
                .json("file:///Users/anuragharsh/Desktop/Data/Activity_Data/")
                .select(
                        // 处理毫秒级Arrival_Time：转秒后用from_unixtime
                        from_unixtime(col("Arrival_Time").divide(1000), "MM-dd-yyyy HH:mm:ss").as("timestamp_1"),
                        // 处理毫秒级Arrival_Time：用to_timestamp+自定义格式
                        date_format(to_timestamp(col("Arrival_Time").divide(1000)), "MM-dd-yyyy HH:mm:ss").as("timestamp_1_alt"),
                        // 处理纳秒级Creation_Time：转秒后用from_unixtime
                        from_unixtime(col("Creation_Time").divide(1000000000), "MM-dd-yyyy HH:mm:ss.SSS").as("timestamp_2"),
                        // 处理纳秒级Creation_Time：转毫秒后用to_timestamp
                        date_format(to_timestamp(col("Creation_Time").divide(1000000)), "MM-dd-yyyy HH:mm:ss.SSS").as("timestamp_2_alt")
                );

        // 流式输出到控制台
        jsondf.writeStream()
                .outputMode("append")
                .format("console")
                .start()
                .awaitTermination();
    }
}

关键说明 #

• Spark列操作不支持直接用/运算符，必须使用divide方法完成数值转换。
• to_timestamp返回Timestamp类型，若需要自定义格式，用date_format包裹即可。
• 纳秒转时间时，若要保留毫秒精度，可将格式调整为MM-dd-yyyy HH:mm:ss.SSS。

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