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字段？请提供一些将resultSchema中的timestamp转换为DateTime的指针。数据类型。

Answer 1

在Beam (or in Calcite)中没有现成的方法可以做到这一点。简短版本-方解石或光束无法知道您实际如何存储整数中的日期或时间戳。然而，假设你有epoch millis，这应该是可行的：

@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，只需在SqlTransform的输出中应用一个自定义的ParDo即可。在该ParDo中，从Row对象提取整数时间戳，将其转换为DateTime，然后发出它，例如，作为具有不同模式的另一行的一部分。