Spark SQL: конвертировать метку времени в миллисекундах в формате JSON в формат даты - PullRequest
Новая
       4

Spark SQL: конвертировать метку времени в миллисекундах в формате JSON в формат даты

0 голосов
/ 13 ноября 2018

Схема имеет dataType, объявленный как Timestamp, но задание spark не преобразует его в нужный формат. 

Dataset<Row> stream = sparkSession.readStream()
          .format("kafka")
          .option("kafka.bootstrap.servers", kafkaBootstrapServersString)
          .option("subscribe", topic)
//          .option("maxOffsetsPerTrigger", 10000)
          .load();

      Dataset<Row> rawStream = stream
              .selectExpr("CAST(value AS STRING)")
              .select(from_json(col("value"), eventSpecificStructType).as("eventData"))
              .select("eventData.*")
              .filter(col("eventType").equalTo("Test"));

Метка времени, поступающая как 1542126896113, преобразуется в 50838-01-28 18:49:111.0.
Есть ли способ привести миллисекунды в формат даты и времени?

Ответы [ 2 ]

0 голосов
/ 23 ноября 2018

Вам нужно будет создать UDF на Java. 

import java.sql.Timestamp;
import java.text.SimpleDateFormat;

SimpleDateFormat dateFormat = new SimpleDateFormat("....Date time pattern...");
spark.udf().register("timestamp", new UDF1<String, Timestamp>() {
    private static final long serialVersionUID = 1335972766810808134L;
    @Override
    public Timestamp call(String source)
    {
      try{
            return new Timestamp(dateFormat.parse(source).getTime());
         } catch (ParseException e) {
                 e.printStackTrace();
         }
      }
      return null;
     }
 }, DataTypes.TimestampType);

Наконец: 

stream = stream.withColumn("col", callUDF("timestamp", dataframe.col("col")));
0 голосов
/ 14 ноября 2018

Как насчет деления значения в миллисекундах на 1000. Является ли ниже ваше ожидание? 

val df = Seq(("1542126896113"),("1542126896116")).toDF("unixtime")
  df.withColumn("times",from_unixtime('unixtime.cast("long")/1000)).show(false)

выход 

+-------------+-------------------+
|unixtime     |times              |
+-------------+-------------------+
|1542126896113|2018-11-13 22:04:56|
|1542126896116|2018-11-13 22:04:56|
+-------------+-------------------+
...