Spark: отметка времени изменяется при чтении из записанного файла

Question

У меня есть паркет, который выглядит следующим образом:

------------ 
name | age | 
------------
Tom  | 12  | 
------------
Mary | 15  | 

Теперь я добавил к нему столбец "метка времени", используя:

final DataFrame dfWithNewColumn = df.withColumn("timestamp", createTimestamp())

, и он выглядит так:

------------ --------------
name | age | timestamp     |
---------------------------
Tom  | 12  | 1569312845998 |
---------------------------
Mary | 15  | 1569312845998 |

И я записываю это в паркет:

    dfWithNewColumn.write()
            .partitionBy(new String[]{"name","timestamp"})
            .mode(SaveMode.Append)
            .parquet(parquetPath);

Когда я смотрю с помощью spark-shell, он в хорошем формате:

------------ --------------
name | age | timestamp     |
---------------------------
Tom  | 12  | 1569312845998 |
---------------------------
Mary | 15  | 1569312845998 |

Но проблема в том,когда я снова читаю паркет:

public static StructType createSchema() {
    final StructType schema = DataTypes.createStructType(Arrays.asList(
            DataTypes.createStructField("age", DataTypes.StringType, false),
            DataTypes.createStructField("name", DataTypes.StringType, false),
            DataTypes.createStructField("timestamp", DataTypes.LongType, false)
    ));
    return schema;
}


    DataFrame df = sqlContext.read()
            .schema(createSchema())
            .parquet(parquetPath);

Когда я показываю строки df.show(), он становится:

------------ --------------
age | name | timestamp     |
---------------------------
12  | Tom  | 171798691853  |
---------------------------
15  | Mary  | 171798691853  |

Как это возможно?С паркетным файлом все в порядке, поэтому я предполагаю, что проблема в коде чтения.

Редактировать:

Я нашел причину.Эта проблема возникает после того, как я изменил spark.sql.sources.partitionColumnTypeInference.enabled = false.Как я могу справиться с этим?

Answer 1

Используйте Spark из коробки функции current_timestamp (), которая возвращает текущую метку времени в качестве столбца метки времени.И во время чтения он должен читаться как org.apache.spark.sql.types.TimestampType datatype.

//Write 
final DataFrame dfWithNewColumn = df.withColumn("timestamp", current_timestamp())
//Read
public static StructType createSchema() {
final StructType schema = DataTypes.createStructType(Arrays.asList(
        DataTypes.createStructField("age", DataTypes.StringType, false),
        DataTypes.createStructField("name", DataTypes.StringType, false),
        DataTypes.createStructField("timestamp", DataTypes.TimestampType, false)
  ));
 return schema;
}