正如这些spark bug报告（link，link）所述，尽管@moemars和@daniel给出了答案，但最新的spark版本（编写本文时为3.0.0和2.4.6）并不完全/正确地支持为所有操作设置时区。
我建议尽可能避免spark中的时间操作，或者在从spark中提取后自己执行，或者使用udf执行，就像这个问题中使用的那样。

改变你的系统时区，并检查它我希望它会工作

在某些情况下，您还需要设置jvm时区。例如，当将数据加载到timestamptype列时，它将解释本地jvm时区中的字符串。要设置jvm时区，您需要为驱动程序和执行程序添加额外的jvm选项：

spark = pyspark.sql.SparkSession \
    .Builder()\
    .appName('test') \
    .master('local') \
    .config('spark.driver.extraJavaOptions', '-Duser.timezone=GMT') \
    .config('spark.executor.extraJavaOptions', '-Duser.timezone=GMT') \
    .config('spark.sql.session.timeZone', 'UTC') \
    .getOrCreate()

我们在本地单元测试环境中执行此操作，因为本地时间不是gmt。
有用的参考：https://en.wikipedia.org/wiki/list_of_tz_database_time_zones

现在您可以使用：

spark.conf.set("spark.sql.session.timeZone", "UTC")

自https://issues.apache.org/jira/browse/spark-18936 在2.2.0中
编辑：
此外，我还将默认时区设置为utc，以避免隐式转换

TimeZone.setDefault(TimeZone.getTimeZone("UTC"))

否则，当要转换的时间戳中没有时区信息时，您将获得从默认时区到utc的隐式转换
例子：

val rawJson = """ {"some_date_field": "2018-09-14 16:05:37"} """

val dsRaw = sparkJob.spark.createDataset(Seq(rawJson))

val output =
  dsRaw
    .select(
      from_json(
        col("value"),
        new StructType(
          Array(
            StructField("some_date_field", DataTypes.TimestampType)
          )
        )
      ).as("parsed")
    ).select("parsed.*")

如果我的默认时区是欧洲/都柏林（gmt+1），并且spark sql会话时区设置为utc，spark将假定“2018-09-14 16:05:37”位于欧洲/都柏林时区，并进行转换（结果将是“2018-09-14 15:05:37”）