Apache calcite: приведение целого числа к дате и времени

Question

Я использую Beam SQL и пытаюсь привести целое число к полю даты и времени.

  Schema resultSchema =
    Schema.builder()
          .addInt64Field("detectedCount")
          .addStringField("sensor")
          .addInt64Field("timestamp")
          .build();

  PCollection<Row> sensorRawUnboundedTimestampedSubset = 
    sensorRowUnbounded.apply(
        SqlTransform.query(
          "select PCOLLECTION.payload.`value`.`count` detectedCount, \n"
          + "PCOLLECTION.payload.`value`.`id` sensor, \n"
          + "PCOLLECTION.`timestamp` `timestamp` \n"
          + "from PCOLLECTION "))
    .setRowSchema(resultSchema);

Для некоторых вычислений и окон, я хочу преобразовать / привести timestamp в Datetime поле? Пожалуйста, предоставьте несколько указателей для конвертации timestamp в resultSchema в DateTime, тип данных.

Answer 1

В Beam ( или в Calcite) нет готового способа сделать это. Короткая версия - у Calcite или Beam нет возможности узнать, как вы на самом деле храните даты или временные метки в целых числах. Однако, если у вас эпоха миллис, это должно сработать:

@Test
public void testBlah() throws Exception {
  // input schema, has timestamps as epoch millis
  Schema schema = Schema.builder().addInt64Field("ts").addStringField("st").build();

  DateTime ts1 = new DateTime(2019, 8, 9, 10, 11, 12);
  DateTime ts2 = new DateTime(2019, 8, 9, 10, 11, 12);

  PCollection<Row> input =
    pipeline
      .apply(
          "createRows",
          Create.of(
              Row.withSchema(schema).addValues(ts1.getMillis(), "two").build(),
              Row.withSchema(schema).addValues(ts2.getMillis(), "twelve").build()))
      .setRowSchema(schema);

  PCollection<Row> result =
    input.apply(
      SqlTransform.query(
          "SELECT \n"
          + "(TIMESTAMP '1970-01-01 00:00:00' + ts * INTERVAL '0.001' SECOND) as ts, \n"
          + "st \n"
          + "FROM \n"
          + "PCOLLECTION"));

  // output schema, has timestamps as DateTime
  Schema outSchema = Schema.builder().addDateTimeField("ts").addStringField("st").build();
  PAssert.that(result)
    .containsInAnyOrder(
        Row.withSchema(outSchema).addValues(ts1, "two").build(),
        Row.withSchema(outSchema).addValues(ts2, "twelve").build());
  pipeline.run();
}

Кроме того, вы всегда можете сделать это в Java, а не в SQL, просто применить пользовательский ParDo на выходе из SqlTransform, В этом ParDoизвлечь целую временную метку изRowобъект, преобразовать его вDateTime а затем испустить его, например, как часть другой строки с другой схемой.