Spark Structured Streaming 自动将时间戳转换为本地时间
Spark Strutured Streaming automatically converts timestamp to local time
我有 UTC 和 ISO8601 时间戳，但使用结构化流，它会自动转换为本地时间。有没有办法停止这种转换？我想在 UTC 中使用它。
我正在从 Kafka 读取 json 数据，然后使用 from_json Spark 函数解析它们。
输入：
{"Timestamp":"2015-01-01T00:00:06.222Z"}
流量：
SparkSession
.builder()
.master("local[*]")
.appName("my-app")
.getOrCreate()
.readStream()
.format("kafka")
... //some magic
.writeStream()
.format("console")
.start()
.awaitTermination();
架构：
StructType schema = DataTypes.createStructType(new StructField[] {
DataTypes.createStructField("Timestamp", DataTypes.TimestampType, true),});
输出：
+--------------------+
| Timestamp|
+--------------------+
|2015-01-01 01:00:...|
|2015-01-01 01:00:...|
+--------------------+
如您所见，小时已自行增加。
PS：我尝试使用 from_utc_timestamp Spark 函数进行试验，但没有成功。
注意：
这个答案在 Spark < 2.2 中主要有用。对于较新的 Spark 版本，请参阅 by astro-asz
但是我们应该注意，从 Spark 2.4.0 开始，spark.sql.session.timeZone 不会设置 user.timezone (java.util.TimeZone.getDefault)。因此单独设置 spark.sql.session.timeZone 会导致相当尴尬的情况，其中 SQL 和非 SQL 组件使用不同的时区设置。
因此我仍然建议明确设置 user.timezone，即使设置了 spark.sql.session.timeZone。
TL;DR 不幸的是，这就是 Spark 现在处理时间戳的方式，实际上没有内置的替代方法，除了直接在纪元时间上操作，而不使用 date/time 公用事业。
您可以在 Spark 开发者列表上进行有见地的讨论：SQL TIMESTAMP semantics vs. SPARK-18350
到目前为止，我发现的最干净的解决方法是将驱动程序和执行程序的 -Duser.timezone 设置为 UTC。例如提交：
bin/spark-shell --conf "spark.driver.extraJavaOptions=-Duser.timezone=UTC" \
--conf "spark.executor.extraJavaOptions=-Duser.timezone=UTC"
或通过调整配置文件(spark-defaults.conf):
spark.driver.extraJavaOptions -Duser.timezone=UTC
spark.executor.extraJavaOptions -Duser.timezone=UTC
对我来说，它可以使用：
spark.conf.set("spark.sql.session.timeZone", "UTC")
它告诉 spark SQL 使用 UTC 作为时间戳的默认时区。例如，我在 spark SQL 中使用了它：
select *, cast('2017-01-01 10:10:10' as timestamp) from someTable
我知道它在 2.0.1 中不起作用。但适用于 Spark 2.2。我也在 SQLTransformer 中使用过并且有效。
虽然我不确定流式传输。
虽然提供了两个非常好的答案，但我发现它们对于解决问题来说都有些沉重。我不想要任何需要修改整个应用程序的时区解析行为的东西，或者会改变我的 JVM 默认时区的方法。历经千辛万苦，我确实找到了解决方案，我将在下面分享...
将时间[/date] 字符串解析为时间戳以进行日期操作，然后正确呈现结果
首先，让我们解决如何让 Spark SQL 正确地将日期[/时间] 字符串（给定格式）解析为时间戳，然后正确地渲染该时间戳的问题，以便它显示与原始字符串输入相同的日期[/时间]。一般做法是：
- convert a date[/time] string to time stamp [via to_timestamp]
[ to_timestamp seems to assume the date[/time] string represents a time relative to UTC (GMT time zone) ]
- relativize that timestamp to the timezone we are in via from_utc_timestamp
下面的测试代码实现了这种方法。 'timezone we are in' 作为第一个参数传递给 timeTricks 方法。该代码将输入字符串“1970-01-01”转换为 localizedTimeStamp（通过 from_utc_timestamp）并验证该时间戳的 'valueOf' 是否与“1970-01-01 00:00:00 相同]".
object TimeTravails {
def main(args: Array[String]): Unit = {
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
val spark: SparkSession = SparkSession.builder()
.master("local[3]")
.appName("SparkByExample")
.getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
import spark.implicits._
import java.sql.Timestamp
def timeTricks(timezone: String): Unit = {
val df2 = List("1970-01-01").toDF("timestr"). // can use to_timestamp even without time parts !
withColumn("timestamp", to_timestamp('timestr, "yyyy-MM-dd")).
withColumn("localizedTimestamp", from_utc_timestamp('timestamp, timezone)).
withColumn("weekday", date_format($"localizedTimestamp", "EEEE"))
val row = df2.first()
println("with timezone: " + timezone)
df2.show()
val (timestamp, weekday) = (row.getAs[Timestamp]("localizedTimestamp"), row.getAs[String]("weekday"))
timezone match {
case "UTC" =>
assert(timestamp == Timestamp.valueOf("1970-01-01 00:00:00") && weekday == "Thursday")
case "PST" | "GMT-8" | "America/Los_Angeles" =>
assert(timestamp == Timestamp.valueOf("1969-12-31 16:00:00") && weekday == "Wednesday")
case "Asia/Tokyo" =>
assert(timestamp == Timestamp.valueOf("1970-01-01 09:00:00") && weekday == "Thursday")
}
}
timeTricks("UTC")
timeTricks("PST")
timeTricks("GMT-8")
timeTricks("Asia/Tokyo")
timeTricks("America/Los_Angeles")
}
}
Structured Streaming Interpreting incoming date[/time] strings as UTC (not local time)问题的解决方案
下面的代码说明了如何应用上述技巧（稍作修改）来纠正时间戳因本地时间和 GMT 之间的偏移而偏移的问题。
object Struct {
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
def main(args: Array[String]): Unit = {
val timezone = "PST"
val spark: SparkSession = SparkSession.builder()
.master("local[3]")
.appName("SparkByExample")
.getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
val df = spark.readStream
.format("socket")
.option("host", "localhost")
.option("port", "9999")
.load()
import spark.implicits._
val splitDf = df.select(split(df("value"), " ").as("arr")).
select($"arr" (0).as("tsString"), $"arr" (1).as("count")).
withColumn("timestamp", to_timestamp($"tsString", "yyyy-MM-dd"))
val grouped = splitDf.groupBy(window($"timestamp", "1 day", "1 day").as("date_window")).count()
val tunedForDisplay =
grouped.
withColumn("windowStart", to_utc_timestamp($"date_window.start", timezone)).
withColumn("windowEnd", to_utc_timestamp($"date_window.end", timezone))
tunedForDisplay.writeStream
.format("console")
.outputMode("update")
.option("truncate", false)
.start()
.awaitTermination()
}
}
代码需要通过套接字输入...我使用程序 'nc' (net cat) 是这样启动的：
nc -l 9999
然后启动Spark程序，给net cat输入一行：
1970-01-01 4
我得到的输出说明了偏移偏移的问题：
-------------------------------------------
Batch: 1
-------------------------------------------
+------------------------------------------+-----+-------------------+-------------------+
|date_window |count|windowStart |windowEnd |
+------------------------------------------+-----+-------------------+-------------------+
|[1969-12-31 16:00:00, 1970-01-01 16:00:00]|1 |1970-01-01 00:00:00|1970-01-02 00:00:00|
+------------------------------------------+-----+-------------------+-------------------+
请注意，date_window 的开始和结束与输入相差八小时（因为我处于 GMT-7/8 时区，太平洋标准时间）。但是，我使用 to_utc_timestamp 更正了这一转变，以获得包含输入的一天 window 的正确开始和结束日期时间：1970-01-01 00:00:00,1970-01- 0200:00:00.
请注意，在第一个代码块中，我们使用了 from_utc_timestamp，而对于结构化流解决方案，我们使用了 to_utc_timestamp。我还没有弄清楚在给定情况下使用这两个中的哪一个。 （如果你知道，请告诉我！）。
另一个对我有用的解决方案是将 jvm 默认时区设置为您的目标时区（在您的情况下为 UTC）。
TimeZone.setDefault(TimeZone.getTimeZone("UTC"));
我在将我的 spark 数据帧写入数据库之前添加了上面的代码。
我有 UTC 和 ISO8601 时间戳，但使用结构化流，它会自动转换为本地时间。有没有办法停止这种转换？我想在 UTC 中使用它。
我正在从 Kafka 读取 json 数据，然后使用 from_json Spark 函数解析它们。
输入：
{"Timestamp":"2015-01-01T00:00:06.222Z"}
流量：
SparkSession
.builder()
.master("local[*]")
.appName("my-app")
.getOrCreate()
.readStream()
.format("kafka")
... //some magic
.writeStream()
.format("console")
.start()
.awaitTermination();
架构：
StructType schema = DataTypes.createStructType(new StructField[] {
DataTypes.createStructField("Timestamp", DataTypes.TimestampType, true),});
输出：
+--------------------+
| Timestamp|
+--------------------+
|2015-01-01 01:00:...|
|2015-01-01 01:00:...|
+--------------------+
如您所见，小时已自行增加。
PS：我尝试使用 from_utc_timestamp Spark 函数进行试验，但没有成功。
注意：
这个答案在 Spark < 2.2 中主要有用。对于较新的 Spark 版本，请参阅
但是我们应该注意，从 Spark 2.4.0 开始，spark.sql.session.timeZone 不会设置 user.timezone (java.util.TimeZone.getDefault)。因此单独设置 spark.sql.session.timeZone 会导致相当尴尬的情况，其中 SQL 和非 SQL 组件使用不同的时区设置。
因此我仍然建议明确设置 user.timezone，即使设置了 spark.sql.session.timeZone。
TL;DR 不幸的是，这就是 Spark 现在处理时间戳的方式，实际上没有内置的替代方法，除了直接在纪元时间上操作，而不使用 date/time 公用事业。
您可以在 Spark 开发者列表上进行有见地的讨论：SQL TIMESTAMP semantics vs. SPARK-18350
到目前为止，我发现的最干净的解决方法是将驱动程序和执行程序的 -Duser.timezone 设置为 UTC。例如提交：
bin/spark-shell --conf "spark.driver.extraJavaOptions=-Duser.timezone=UTC" \
--conf "spark.executor.extraJavaOptions=-Duser.timezone=UTC"
或通过调整配置文件(spark-defaults.conf):
spark.driver.extraJavaOptions -Duser.timezone=UTC
spark.executor.extraJavaOptions -Duser.timezone=UTC
对我来说，它可以使用：
spark.conf.set("spark.sql.session.timeZone", "UTC")
它告诉 spark SQL 使用 UTC 作为时间戳的默认时区。例如，我在 spark SQL 中使用了它：
select *, cast('2017-01-01 10:10:10' as timestamp) from someTable
我知道它在 2.0.1 中不起作用。但适用于 Spark 2.2。我也在 SQLTransformer 中使用过并且有效。
虽然我不确定流式传输。
虽然提供了两个非常好的答案，但我发现它们对于解决问题来说都有些沉重。我不想要任何需要修改整个应用程序的时区解析行为的东西，或者会改变我的 JVM 默认时区的方法。历经千辛万苦，我确实找到了解决方案，我将在下面分享...
将时间[/date] 字符串解析为时间戳以进行日期操作，然后正确呈现结果
首先，让我们解决如何让 Spark SQL 正确地将日期[/时间] 字符串（给定格式）解析为时间戳，然后正确地渲染该时间戳的问题，以便它显示与原始字符串输入相同的日期[/时间]。一般做法是：
- convert a date[/time] string to time stamp [via to_timestamp]
[ to_timestamp seems to assume the date[/time] string represents a time relative to UTC (GMT time zone) ]
- relativize that timestamp to the timezone we are in via from_utc_timestamp
下面的测试代码实现了这种方法。 'timezone we are in' 作为第一个参数传递给 timeTricks 方法。该代码将输入字符串“1970-01-01”转换为 localizedTimeStamp（通过 from_utc_timestamp）并验证该时间戳的 'valueOf' 是否与“1970-01-01 00:00:00 相同]".
object TimeTravails {
def main(args: Array[String]): Unit = {
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
val spark: SparkSession = SparkSession.builder()
.master("local[3]")
.appName("SparkByExample")
.getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
import spark.implicits._
import java.sql.Timestamp
def timeTricks(timezone: String): Unit = {
val df2 = List("1970-01-01").toDF("timestr"). // can use to_timestamp even without time parts !
withColumn("timestamp", to_timestamp('timestr, "yyyy-MM-dd")).
withColumn("localizedTimestamp", from_utc_timestamp('timestamp, timezone)).
withColumn("weekday", date_format($"localizedTimestamp", "EEEE"))
val row = df2.first()
println("with timezone: " + timezone)
df2.show()
val (timestamp, weekday) = (row.getAs[Timestamp]("localizedTimestamp"), row.getAs[String]("weekday"))
timezone match {
case "UTC" =>
assert(timestamp == Timestamp.valueOf("1970-01-01 00:00:00") && weekday == "Thursday")
case "PST" | "GMT-8" | "America/Los_Angeles" =>
assert(timestamp == Timestamp.valueOf("1969-12-31 16:00:00") && weekday == "Wednesday")
case "Asia/Tokyo" =>
assert(timestamp == Timestamp.valueOf("1970-01-01 09:00:00") && weekday == "Thursday")
}
}
timeTricks("UTC")
timeTricks("PST")
timeTricks("GMT-8")
timeTricks("Asia/Tokyo")
timeTricks("America/Los_Angeles")
}
}
Structured Streaming Interpreting incoming date[/time] strings as UTC (not local time)问题的解决方案
下面的代码说明了如何应用上述技巧（稍作修改）来纠正时间戳因本地时间和 GMT 之间的偏移而偏移的问题。
object Struct {
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
def main(args: Array[String]): Unit = {
val timezone = "PST"
val spark: SparkSession = SparkSession.builder()
.master("local[3]")
.appName("SparkByExample")
.getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
val df = spark.readStream
.format("socket")
.option("host", "localhost")
.option("port", "9999")
.load()
import spark.implicits._
val splitDf = df.select(split(df("value"), " ").as("arr")).
select($"arr" (0).as("tsString"), $"arr" (1).as("count")).
withColumn("timestamp", to_timestamp($"tsString", "yyyy-MM-dd"))
val grouped = splitDf.groupBy(window($"timestamp", "1 day", "1 day").as("date_window")).count()
val tunedForDisplay =
grouped.
withColumn("windowStart", to_utc_timestamp($"date_window.start", timezone)).
withColumn("windowEnd", to_utc_timestamp($"date_window.end", timezone))
tunedForDisplay.writeStream
.format("console")
.outputMode("update")
.option("truncate", false)
.start()
.awaitTermination()
}
}
代码需要通过套接字输入...我使用程序 'nc' (net cat) 是这样启动的：
nc -l 9999
然后启动Spark程序，给net cat输入一行：
1970-01-01 4
我得到的输出说明了偏移偏移的问题：
-------------------------------------------
Batch: 1
-------------------------------------------
+------------------------------------------+-----+-------------------+-------------------+
|date_window |count|windowStart |windowEnd |
+------------------------------------------+-----+-------------------+-------------------+
|[1969-12-31 16:00:00, 1970-01-01 16:00:00]|1 |1970-01-01 00:00:00|1970-01-02 00:00:00|
+------------------------------------------+-----+-------------------+-------------------+
请注意，date_window 的开始和结束与输入相差八小时（因为我处于 GMT-7/8 时区，太平洋标准时间）。但是，我使用 to_utc_timestamp 更正了这一转变，以获得包含输入的一天 window 的正确开始和结束日期时间：1970-01-01 00:00:00,1970-01- 0200:00:00.
请注意，在第一个代码块中，我们使用了 from_utc_timestamp，而对于结构化流解决方案，我们使用了 to_utc_timestamp。我还没有弄清楚在给定情况下使用这两个中的哪一个。 （如果你知道，请告诉我！）。
另一个对我有用的解决方案是将 jvm 默认时区设置为您的目标时区（在您的情况下为 UTC）。
TimeZone.setDefault(TimeZone.getTimeZone("UTC"));
我在将我的 spark 数据帧写入数据库之前添加了上面的代码。