Python PySpark SQL 中的日期时间范围过滤器
声明：本页面是StackOverFlow热门问题的中英对照翻译，遵循CC BY-SA 4.0协议，如果您需要使用它，必须同样遵循CC BY-SA许可，注明原文地址和作者信息，同时你必须将它归于原作者（不是我）：StackOverFlow
原文地址: http://stackoverflow.com/questions/31407461/
Warning: these are provided under cc-by-sa 4.0 license. You are free to use/share it, But you must attribute it to the original authors (not me):
StackOverFlow
datetime range filter in PySpark SQL
提问by funkifunki
What is the correct way to filter data frame by timestamp field?
按时间戳字段过滤数据帧的正确方法是什么？
I have tried different date formats and forms of filtering, nothing helps: either pyspark returns 0 objects, or throws an error that it doesn't understand datetime format
我尝试了不同的日期格式和过滤形式，没有任何帮助：pyspark 返回 0 个对象，或者抛出它不理解日期时间格式的错误
Here is what i got so far:
这是我到目前为止所得到的：
from pyspark import SparkContext
from pyspark.sql import SQLContext
from django.utils import timezone
from django.conf import settings
from myapp.models import Collection
sc = SparkContext("local", "DjangoApp")
sqlc = SQLContext(sc)
url = "jdbc:postgresql://%(HOST)s/%(NAME)s?user=%(USER)s&password=%(PASSWORD)s" % settings.DATABASES['default']
sf = sqlc.load(source="jdbc", url=url, dbtable='myapp_collection')
range for timestamp field:
时间戳字段的范围：
system_tz = timezone.pytz.timezone(settings.TIME_ZONE)
date_from = datetime.datetime(2014, 4, 16, 18, 30, 0, 0, tzinfo=system_tz)
date_to = datetime.datetime(2015, 6, 15, 18, 11, 59, 999999, tzinfo=system_tz)
attempt 1
尝试 1
date_filter = "my_col >= '%s' AND my_col <= '%s'" % (
date_from.isoformat(), date_to.isoformat()
)
sf = sf.filter(date_filter)
sf.count()
Out[12]: 0
attempt 2
尝试 2
sf = sf.filter(sf.my_col >= date_from).filter(sf.my_col <= date_to)
sf.count()
---------------------------------------------------------------------------
Py4JJavaError: An error occurred while calling o63.count.
: org.apache.spark.SparkException: Job aborted due to stage failure:
Task 0 in stage 4.0 failed 1 times, most recent failure:
Lost task 0.0 in stage 4.0 (TID 3, localhost): org.postgresql.util.PSQLException:
ERROR: syntax error at or near "18"
#
# ups.. JDBC doesn't understand 24h time format??
attempt 3
尝试 3
sf = sf.filter("my_col BETWEEN '%s' AND '%s'" % \
(date_from.isoformat(), date_to.isoformat())
)
---------------------------------------------------------------------------
Py4JJavaError: An error occurred while calling o97.count.
: org.apache.spark.SparkException: Job aborted due to stage failure:
Task 0 in stage 17.0 failed 1 times, most recent failure:
Lost task 0.0 in stage 17.0 (TID 13, localhost): org.postgresql.util.PSQLException:
ERROR: syntax error at or near "18"
the data do exist in the table, though:
数据确实存在于表中，但：
django_filters = {
'my_col__gte': date_from,
'my_col__lte': date_to
}
Collection.objects.filter(**django_filters).count()
Out[17]: 1093436
Or this way
或者这样
django_range_filter = {'my_col__range': (date_from, date_to)}
Collection.objects.filter(**django_range_filter).count()
Out[19]: 1093436
采纳答案by zero323
Lets assume your data frame looks as follows:
让我们假设您的数据框如下所示：
sf = sqlContext.createDataFrame([
[datetime.datetime(2013, 6, 29, 11, 34, 29)],
[datetime.datetime(2015, 7, 14, 11, 34, 27)],
[datetime.datetime(2012, 3, 10, 19, 00, 11)],
[datetime.datetime(2016, 2, 8, 12, 21)],
[datetime.datetime(2014, 4, 4, 11, 28, 29)]
], ('my_col', ))
with schema:
使用架构：
root
|-- my_col: timestamp (nullable = true)
and you want to find dates in a following range:
并且您想在以下范围内查找日期：
import datetime, time
dates = ("2013-01-01 00:00:00", "2015-07-01 00:00:00")
timestamps = (
time.mktime(datetime.datetime.strptime(s, "%Y-%m-%d %H:%M:%S").timetuple())
for s in dates)
It is possible to query using timestamps either computed on a driver side:
可以使用在驱动程序端计算的时间戳进行查询：
q1 = "CAST(my_col AS INT) BETWEEN {0} AND {1}".format(*timestamps)
sf.where(q1).show()
or using unix_timestampfunction:
或使用unix_timestamp功能：
q2 = """CAST(my_col AS INT)
BETWEEN unix_timestamp('{0}', 'yyyy-MM-dd HH:mm:ss')
AND unix_timestamp('{1}', 'yyyy-MM-dd HH:mm:ss')""".format(*dates)
sf.where(q2).show()
It is also possible to use udf in a similar way I described in an another answer.
也可以按照我在另一个答案中描述的类似方式使用 udf 。
If you use raw SQL it is possible to extract different elements of timestamp using year, date, etc.
如果使用原始SQL，可以使用提取时间戳的不同元素year，date等等。
sqlContext.sql("""SELECT * FROM sf
WHERE YEAR(my_col) BETWEEN 2014 AND 2015").show()
EDIT:
编辑：
Since Spark 1.5 you can use built-in functions:
从 Spark 1.5 开始，您可以使用内置函数：
dates = ("2013-01-01", "2015-07-01")
date_from, date_to = [to_date(lit(s)).cast(TimestampType()) for s in dates]
sf.where((sf.my_col > date_from) & (sf.my_col < date_to))
You can also use pyspark.sql.Column.between, which is inclusive of the bounds:
您还可以使用pyspark.sql.Column.between，它包含边界：
from pyspark.sql.functions import col
sf.where(col('my_col').between(*dates)).show(truncate=False)
#+---------------------+
#|my_col |
#+---------------------+
#|2013-06-29 11:34:29.0|
#|2014-04-04 11:28:29.0|
#+---------------------+
回答by Sean
How about something like this:
这样的事情怎么样：
import pyspark.sql.functions as func
df = df.select(func.to_date(df.my_col).alias("time"))
sf = df.filter(df.time > date_from).filter(df.time < date_to)
回答by alofgran
The following seems to be working for me (someone let me know if this is bad form or inaccurate though)...
以下似乎对我有用（有人告诉我这是错误的形式还是不准确的）......
First, create a new column for each end of the window (in this example, it's 100 days to 200 days after the date in column: column_name.
首先，为窗口的每一端创建一个新列（在本例中，列中的日期之后 100 天到 200 天：column_name。
from pyspark.sql import functions as F
new_df = new_df.withColumn('After100Days', F.lit(F.date_add(new_df['column_name'], 100)))
new_df = new_df.withColumn('After200Days', F.lit(F.date_add(new_df['column_name'], 200)))
Filter as follows...
过滤如下...
For filtering dates inside a particular range:
过滤特定范围内的日期：
result= df.where((df.col1> df.col2) & (df.col1 < df.col3))
For filtering dates outside a particular range:
过滤特定范围之外的日期：
result= df.where((df.col1 < df.col2) | (df.col1 > df.col3))
