Spark Strutured Streaming автоматически преобразует временную метку в локальное время

Ответ 1

Для меня это работало:

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

Он сообщает, что искра SQL использует UTC в качестве часового пояса по умолчанию для временных меток. Я использовал его в искрах SQL, например:

select *, cast('2017-01-01 10:10:10' as timestamp) from someTable

Я знаю, что он не работает в версии 2.0.1. но работает в Spark 2.2. Я тоже использовал SQLTransformer и он сработал.

Однако я не уверен в потоковой передаче.

Ответ 2

Примечание:

Этот ответ полезен в основном в Spark & lt; 2.2. Для более новой версии Spark см. ответ 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 обрабатывает метки времени прямо сейчас, и на самом деле нет никакой встроенной альтернативы, кроме прямой работы в эпоху, без использования утилит даты/времени.

Вы можете подробно обсудить список разработчиков Spark: Семантика SQL TIMESTAMP и 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

Ответ 3

Хотя были даны два очень хороших ответа, я нашел, что оба они были немного тяжелым молотком для решения проблемы. Я не хотел ничего, что потребовало бы изменения поведения разбора часового пояса во всем приложении, или подход, который изменил бы часовой пояс по умолчанию для моей JVM. Я нашел решение после большой боли, о которой я расскажу ниже...

Разбор строк времени [/date] в метки времени для манипуляций с датой, а затем корректное отображение результата обратно

Во-первых, давайте рассмотрим вопрос о том, как заставить Spark SQL правильно анализировать строку даты [/time] (с учетом формата) в метку времени, а затем правильно отобразить эту метку времени так, чтобы она отображала ту же дату [/time], что и метка времени. ввод исходной строки. Общий подход:

- 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 

Тестовый код ниже реализует этот подход. 'часовой пояс, в котором мы находимся', передается в качестве первого аргумента методу 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")
  }
}

Решение проблемы структурированной потоковой передачи. Интерпретация входящих строк даты [/time] как UTC (не локального времени)

Приведенный ниже код иллюстрирует, как применить вышеуказанные приемы (с небольшой модификацией), чтобы исправить проблему смещения временных меток на смещение между местным временем и 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, PST). Однако я исправляю этот сдвиг, используя to_utc_timestamp, чтобы получить правильное время начала и окончания для однодневного окна, которое включает ввод: 1970-01-01 00: 00: 00,1970-01-02 00:00:00.

Обратите внимание, что в первом представленном блоке кода мы использовали from_utc_timestamp, тогда как для решения структурированной потоковой передачи мы использовали to_utc_timestamp. Мне еще предстоит выяснить, какой из этих двух вариантов использовать в данной ситуации. (Пожалуйста, подскажите мне, если вы знаете!).