如何修复非法 Parquet 类型：INT64 (TIMESTAMP_MICROS) 错误

Question

我每天使用

sqlContext.read.parquet

中的

PySpark

功能来读取

parquet

文件。数据有一个

timestamp

列。他们将时间戳字段从

2019-08-26T00:00:13.600+0000

更改为

2019-08-26T00:00:13.600Z

。它在 Databricks 中读取良好，但当我尝试通过 Spark 集群读取它时，它给出了

Illegal Parquet type: INT64 (TIMESTAMP_MICROS)

错误。如何使用

read.parquet

函数本身阅读这个新专栏？

目前我使用：

from_unixtime(unix_timestamp(ts,"yyyy-MM-dd HH:mm:ss.SSS"),"yyyy-MM-dd")

作为 ts 将

2019-08-26T00:00:13.600+0000

转换为

2019-08-26

格式。

如何将

2019-08-26T00:00:13.600Z

转换为

2019-08-26

？

Answer 1

这是scala版本

import spark.implicits._
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._

val df2 = Seq(("a3fac", "2019-08-26T00:00:13.600Z")).toDF("id", "eventTime")
val df3= df2.withColumn("eventTime1", to_date(unix_timestamp($"eventTime", "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'").cast(TimestampType)))

df3.show(false)
+-----+------------------------+----------+
|id   |eventTime               |eventTime1|
+-----+------------------------+----------+
|a3fac|2019-08-26T00:00:13.600Z|2019-08-26|
+-----+------------------------+----------+

以下行将时区日期转换为日期

to_date(unix_timestamp($"eventTime", "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'").cast(TimestampType))

pyspark版本：

>>> from pyspark.sql.functions import col, to_date,unix_timestamp
>>> df2=spark.createDataFrame([("a3fac", "2019-08-26T00:00:13.600Z")], ['id', 'eventTime'])
>>> df3=df2.withColumn("eventTime1", to_date(unix_timestamp(col("eventTime"), "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'").cast('timestamp')))
>>> df3.show()
+-----+--------------------+----------+
|   id|           eventTime|eventTime1|
+-----+--------------------+----------+
|a3fac|2019-08-26T00:00:...|2019-08-26|
+-----+--------------------+----------+

Answer 2

您可以使用功能模块中的 to_date api

import pyspark.sql.functions as f

dfl2 = spark.createDataFrame([(1, "2019-08-26T00:00:13.600Z"),]).toDF('col1', 'ts')

dfl2.show(1, False)
+----+------------------------+
|col1|ts                      |
+----+------------------------+
|1   |2019-08-26T00:00:13.600Z|
+----+------------------------+

dfl2.withColumn('date',f.to_date('ts', "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'")).show(1, False)

+----+------------------------+----------+
|col1|ts                      |date      |
+----+------------------------+----------+
|1   |2019-08-26T00:00:13.600Z|2019-08-26|
+----+------------------------+----------+

dfl2.withColumn('date',f.to_date('ts', "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'")).printSchema()
root
 |-- col1: long (nullable = true)
 |-- ts: string (nullable = true)
 |-- date: date (nullable = true)

Answer 3

这可能是由于您的数据被一个系统写入镶木地板，而您正在尝试从另一个系统读取镶木地板。这两个系统使用不同版本的 Parquet。

我也有类似的问题。就我而言，我在 Python 中准备了带有时间戳列的数据，并使用

pandas.to_parquet

保存它。后来我尝试在 EMR 上读取 Pyspark 中的 parquet。我的错误是：

org.apache.spark.sql.AnalysisException: Illegal Parquet type: INT64 (TIMESTAMP(NANOS,false))

经过一番查找，发现是Pandas和Pyspark能够处理的时间戳分辨率不同造成的。我的数据集有几个时间戳列。在 Pandas 中，时间戳以纳秒分辨率表示。但是，Pyspark 使用的某些 Parquet 读取器可能仅支持以毫秒（“ms”）或微秒（“us”）分辨率存储的时间戳。请参阅 https://arrow.apache.org/docs/python/parquet.html#storing-timestamps 了解更多讨论。

因此，当我使用时间戳列保存 Pandas 数据帧时，我显式地将时间戳列强制为微秒分辨率。