Конвертировать из long в Timestamp для Вставки в БД

Question

Цель: чтение данных из файла JSON, где отметка времени является длинным типом, и вставка в таблицу с типом отметки времени.Проблема в том, что я не знаю, как преобразовать длинный тип в тип метки времени для вставки.

Пример входного файла:

    {"sensor_id":"sensor1","reading_time":1549533263587,"notes":"My Notes for 
    Sensor1","temperature":24.11,"humidity":42.90}

Я хочу прочитать это, создатьБоб из него и вставьте в стол.Вот мое определение Bean:

public class DummyBean {

    private String sensor_id;
    private String notes;
    private Timestamp reading_time;
    private double temperature;
    private double humidity;

Вот таблица, в которую я хочу вставить:

    create table dummy (
    id serial not null primary key,
    sensor_id   varchar(40),
    notes   varchar(40),
    reading_time    timestamp with time zone default (current_timestamp at time zone 'UTC'),
    temperature   decimal(15,2),
    humidity      decimal(15,2)
    );

Вот мое приложение Spark, которое читает файл JSON и выполняет вставку (добавьте)

SparkSession spark = SparkSession
                .builder()
                .appName("SparkJDBC2")
                .getOrCreate();

        // Java Bean used to apply schema to JSON Data
        Encoder<DummyBean> dummyEncoder = Encoders.bean(DummyBean.class);

        // Read JSON file to DataSet
        String jsonPath = "input/dummy.json";
        Dataset<DummyBean> readings = spark.read().json(jsonPath).as(dummyEncoder);

        // Diagnostics and Sink
        readings.printSchema();
        readings.show();


        // Write to JDBC Sink
        String url = "jdbc:postgresql://dbhost:5432/mydb";
        String table = "dummy";
        Properties connectionProperties = new Properties();
        connectionProperties.setProperty("user", "foo");
        connectionProperties.setProperty("password", "bar");
        readings.write().mode(SaveMode.Append).jdbc(url, table, connectionProperties);

Вывод и сообщение об ошибке:

root
 |-- humidity: double (nullable = true)
 |-- notes: string (nullable = true)
 |-- reading_time: long (nullable = true)
 |-- sensor_id: string (nullable = true)
 |-- temperature: double (nullable = true)

+--------+--------------------+-------------+---------+-----------+
|humidity|               notes| reading_time|sensor_id|temperature|
+--------+--------------------+-------------+---------+-----------+
|    42.9|My Notes for Sensor1|1549533263587|  sensor1|      24.11|
+--------+--------------------+-------------+---------+-----------+

Exception in thread "main" org.apache.spark.sql.AnalysisException: Column "reading_time" not found in schema Some(StructType(StructField(id,IntegerType,false), StructField(sensor_id,StringType,true), StructField(notes,StringType,true), StructField(temperature,DecimalType(15,2),true), StructField(humidity,DecimalType(15,2),true)));
    at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$$anonfun$4$$anonfun$6.apply(JdbcUtils.scala:147)
    at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$$anonfun$4$$anonfun$6.apply(JdbcUtils.scala:147)
    at scala.Option.getOrElse(Option.scala:121)
    at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$$anonfun$4.apply(JdbcUtils.scala:146)

Answer 1

Исключение в вашем посте говорит, что столбец "read_time" не найден .. поэтому, пожалуйста, проверьте, есть ли в таблице необходимый столбец в конце db.Кроме того, отметка времени дается в миллисекундах, поэтому вам нужно разделить ее на 1000, прежде чем применять функцию to_timestamp (), в противном случае вы получите странную дату.

Я могу воспроизвести ниже и преобразоватьвремя чтения.

scala> val readings = Seq((42.9,"My Notes for Sensor1",1549533263587L,"sensor1",24.11)).toDF("humidity","notes","reading_time","sensor_id","temperature")
readings: org.apache.spark.sql.DataFrame = [humidity: double, notes: string ... 3 more fields]

scala> readings.printSchema();
root
 |-- humidity: double (nullable = false)
 |-- notes: string (nullable = true)
 |-- reading_time: long (nullable = false)
 |-- sensor_id: string (nullable = true)
 |-- temperature: double (nullable = false)


scala> readings.show(false)
+--------+--------------------+-------------+---------+-----------+
|humidity|notes               |reading_time |sensor_id|temperature|
+--------+--------------------+-------------+---------+-----------+
|42.9    |My Notes for Sensor1|1549533263587|sensor1  |24.11      |
+--------+--------------------+-------------+---------+-----------+


scala>  readings.withColumn("ts", to_timestamp('reading_time/1000)).show(false)
+--------+--------------------+-------------+---------+-----------+-----------------------+
|humidity|notes               |reading_time |sensor_id|temperature|ts                     |
+--------+--------------------+-------------+---------+-----------+-----------------------+
|42.9    |My Notes for Sensor1|1549533263587|sensor1  |24.11      |2019-02-07 04:54:23.587|
+--------+--------------------+-------------+---------+-----------+-----------------------+


scala>

Answer 2

Спасибо за вашу помощь.Да, в таблице отсутствовал столбец, поэтому я исправил это.Это то, что решил (версия Java)

import static org.apache.spark.sql.functions.col;
import static org.apache.spark.sql.functions.to_timestamp;

...
Dataset<Row>  readingsRow = readings.withColumn("reading_time", to_timestamp(col("reading_time").$div(1000L)));

// Write to JDBC Sink
String url = "jdbc:postgresql://dbhost:5432/mydb";
String table = "dummy";
Properties connectionProperties = new Properties();
connectionProperties.setProperty("user", "foo");
connectionProperties.setProperty("password", "bar");
readingsRow.write().mode(SaveMode.Append).jdbc(url, table, connectionProperties);

Answer 3

Если ваша дата строковая, вы можете использовать

String readtime = obj.getString("reading_time");
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ssZ"); //Z for time zone
Date reading_time = sdf.parse(readtime);

или

new Date(json.getLong(milliseconds))

, если она длинная