фильтр на отметке времени искры не работает в диапазоне, превышающем день

Question

Я работаю с версией Pyspark 2.3.0. я фильтрую фрейм данных в столбце метки времени.

| - requestTs: отметка времени (nullable = true)

, когда я фильтрую по дневному диапазону времени, он прекрасно работает. когда я устанавливаю фильтр на диапазон 2 дня, он не возвращает все записи. я пробовал несколько способов, как:

        df1 = df.filter(df["requestts"] >= sqlf.lit(startDatestr)) \
           .filter(df["requestts"] <= sqlf.lit(endDatestr))

или

        dates = (startDatestr, endDatestr)

        q1 = "unix_timestamp(requestts) BETWEEN unix_timestamp('{0}', 'yyyy-MM-dd HH:mm:ss') AND unix_timestamp('{1}', 'yyyy-MM-dd HH:mm:ss')".format(
            *dates)

        df1 = df.where(q1)

, где

startDatestr: 2018-06-26 07:00:00
endDatestr: 2018-06-26 21:40:00

когда я считаю возвращенные записи, я получаю

after date filter, count is :1168940768

если я расширю поиск (так что я ожидаю большего или равного количества записей), например:

startDatestr: 2018-06-25 11:00:00
endDatestr: 2018-06-26 21:40:00

я получаю гораздо меньше записей.

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

Answer 1

Ваш запрос должен работать. Я только что попробовал это на оболочке pyspark.

>>> from datetime import datetime
>>> import pyspark.sql.functions as F

>>> columns = ['id', 'ts']
>>> vals = [
...     (1, datetime(2018, 6, 26)),
...     (2, datetime(2018, 6, 27)),
...     (3, datetime(2018, 6, 28)),
...     (4, datetime(2018, 6, 29)),
...     (5, datetime(2018, 6, 30))
... ]
>>> df = spark.createDataFrame(vals, columns)

>>> df.show()
+---+-------------------+
| id|                 ts|
+---+-------------------+
|  1|2018-06-26 00:00:00|
|  2|2018-06-27 00:00:00|
|  3|2018-06-28 00:00:00|
|  4|2018-06-29 00:00:00|
|  5|2018-06-30 00:00:00|
+---+-------------------+

>>> df.printSchema()
root
 |-- id: long (nullable = true)
 |-- ts: timestamp (nullable = true)

Вот запрос, похожий на ваш:

>>> df.filter(df['ts'] >= F.lit('2018-06-27 00:00:00')) \
...     .filter(df['ts'] < F.lit('2018-06-29 00:00:00')).show()
+---+-------------------+
| id|                 ts|
+---+-------------------+
|  2|2018-06-27 00:00:00|
|  3|2018-06-28 00:00:00|
+---+-------------------+

Вы также можете использовать логический оператор, чтобы сделать два фильтра одновременно:

>>> df.filter((df['ts'] >= F.lit('2018-06-27 00:00:00'))
...           & (df['ts'] < F.lit('2018-06-29 00:00:00'))).show()
+---+-------------------+
| id|                 ts|
+---+-------------------+
|  2|2018-06-27 00:00:00|
|  3|2018-06-28 00:00:00|
+---+-------------------+

Наконец, вам не нужна функция lit. Так что вы можете просто использовать строку:

>>> df.filter((df['ts'] >= '2018-06-27 00:00:00')
...           & (df['ts'] < '2018-06-29 00:00:00')).show()
+---+-------------------+
| id|                 ts|
+---+-------------------+
|  2|2018-06-27 00:00:00|
|  3|2018-06-28 00:00:00|
+---+-------------------+

Для часового пояса вы можете увидеть документацию pyspark для из функции __unixtime . Я также вставил приведенный ниже пример для удобства:

>>> spark.conf.set("spark.sql.session.timeZone", "America/Los_Angeles")
>>> time_df = spark.createDataFrame([(1428476400,)], ['unix_time'])
>>> time_df.select(from_unixtime('unix_time').alias('ts')).collect()
[Row(ts='2015-04-08 00:00:00')]
>>> spark.conf.unset("spark.sql.session.timeZone")