Parquet int96 时间戳通过 python 转换为 datetime/date

Question

TL;DR我想将int96值（例如ACIE4NxJAAAKhSUA）转换为可读的时间戳格式，例如2020-03-02 14：34：22或任何可以正常解释的格式...我主要使用python，所以我正在寻找一个执行此转换的函数。如果有另一个功能可以反其道而行之 -- 甚至更好。背景我正在使用镶木地板工具通过以下通信将原始镶木地板文件（具有快速压缩）转换为原始JSON：C:\Research> java -jar parquet-tools-1.8.2.jar cat --json original-file.snappy.parquet > parquet-output.json在JSON中，我看到这些值作为时间戳：{... "_id":"101836","timestamp":"ACIE4NxJAAAKhSUA"}我已经确定“ACIE4NxJAAAKhSUA”的时间戳值确实是int96（这也通过读取镶木地板文件的架构来确认....message spark_schema { ...(stuff)...  optional binary _id (UTF8);  optional int96 timestamp;}我认为这也被称为Impala Timestamp（至少这是我收集到的）进一步的问题研究我一直在到处寻找一些关于如何“读取”int96值的函数或信息（进入python - 我想用这种语言保留它，因为我最熟悉它）并输出时间戳 - 我什么也没找到。以下是我已经研究过的一篇文章（与这个主题有关）：ParquetWriter在SO的研究在这里通过golan在SO这里铸造int96注意：这有一个功能，我可以探索，但我不确定如何深入研究关于折旧的 int96 时间戳请不要要求我停止在镶木地板文件中使用旧的/折旧的时间戳格式，我很清楚到目前为止我所做的研究。我是文件/数据的接收者 - 我无法更改创建时使用的格式。如果有另一种方法可以控制初始JSON输出以提供“non int96”值 - 我也会对此感兴趣。非常感谢您对SO社区的帮助！

Answer 1

拼花工具将无法将格式类型从 INT96 更改为 INT64。您在 json 输出中观察到的是存储在 INT96 时间戳类型中的时间戳的字符串表示形式。您需要火花在INT64 TimestampType中使用时间戳重写此镶木地板，然后json输出将生成时间戳（以您想要的格式）。

您需要在 Spark 中设置特定的配置 -

spark-shell --conf spark.sql.parquet.outputTimestampType=TIMESTAMP_MICROS

2020-03-16 11:37:50 WARN  NativeCodeLoader:62 - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable

Setting default log level to "WARN".

To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).

Spark context Web UI available at http://192.168.0.20:4040

Spark context available as 'sc' (master = local[*], app id = local-1584383875924).

Spark session available as 'spark'.

Welcome to

      ____              __

     / __/__  ___ _____/ /__

    _\ \/ _ \/ _ `/ __/  '_/

   /___/ .__/\_,_/_/ /_/\_\   version 2.4.0

      /_/

Using Scala version 2.11.12 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_91)

Type in expressions to have them evaluated.

Type :help for more information.

val sourceDf = spark.read.parquet("original-file.snappy.parquet")

2020-03-16 11:38:31 WARN  Utils:66 - Truncated the string representation of a plan since it was too large. This behavior can be adjusted by setting 'spark.debug.maxToStringFields' in SparkEnv.conf.

sourceDf: org.apache.spark.sql.DataFrame = [application: struct<name: string, upgrades: struct<value: double> ... 3 more fields>, timestamp: timestamp ... 16 more fields]

scala> sourceDf.repartition(1).write.parquet("Downloads/output")

拼花工具将显示正确的时间戳类型

parquet-tools schema Downloads/output/part-00000-edba239b-e696-4b4e-8fd3-c7cca9eea6bf-c000.snappy.parquet 

message spark_schema {

  ...

  optional binary _id (UTF8);

  optional int64 timestamp (TIMESTAMP_MICROS);

  ...

}

而 json 转储给出了 -

parquet-tools cat --json Downloads/output/part-00000-edba239b-e696-4b4e-8fd3-c7cca9eea6bf-c000.snappy.parquet

{..."_id":"101836", "timestamp":1583973827000000}

记录的时间戳以纳秒为单位。希望这有帮助！

Answer 2

Doug，这个来自 arrow/cpp/src/parquet/types.h 的代码显示了 Int96 时间戳是如何在内部存储的：

constexpr int64_t kJulianToUnixEpochDays = INT64_C(2440588);

constexpr int64_t kSecondsPerDay = INT64_C(60 * 60 * 24);

constexpr int64_t kMillisecondsPerDay = kSecondsPerDay * INT64_C(1000);

constexpr int64_t kMicrosecondsPerDay = kMillisecondsPerDay * INT64_C(1000);

constexpr int64_t kNanosecondsPerDay = kMicrosecondsPerDay * INT64_C(1000);

MANUALLY_ALIGNED_STRUCT(1) Int96 { uint32_t value[3]; };

STRUCT_END(Int96, 12);

static inline void Int96SetNanoSeconds(parquet::Int96& i96, int64_t nanoseconds) {

  std::memcpy(&i96.value, &nanoseconds, sizeof(nanoseconds));

}

static inline int64_t Int96GetNanoSeconds(const parquet::Int96& i96) {

  // We do the computations in the unsigned domain to avoid unsigned behaviour

  // on overflow.

  uint64_t days_since_epoch =

      i96.value[2] - static_cast<uint64_t>(kJulianToUnixEpochDays);

  uint64_t nanoseconds = 0;

  memcpy(&nanoseconds, &i96.value, sizeof(uint64_t));

  return static_cast<int64_t>(days_since_epoch * kNanosecondsPerDay + nanoseconds);

}