Python PySpark 中的列过滤
声明：本页面是StackOverFlow热门问题的中英对照翻译，遵循CC BY-SA 4.0协议，如果您需要使用它，必须同样遵循CC BY-SA许可，注明原文地址和作者信息，同时你必须将它归于原作者（不是我）：StackOverFlow
原文地址: http://stackoverflow.com/questions/31400143/
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
Column filtering in PySpark
提问by oikonomiyaki
I have a dataframe dfloaded from Hive table and it has a timestamp column, say ts, with string type of format dd-MMM-yy hh.mm.ss.MS a(converted to python datetime library, this is %d-%b-%y %I.%M.%S.%f %p).
我有一个df从 Hive 表加载的数据框，它有一个时间戳列，例如ts，格式为字符串类型dd-MMM-yy hh.mm.ss.MS a（转换为 python datetime 库，这是%d-%b-%y %I.%M.%S.%f %p）。
Now I want to filter rows from the dataframe that are from the last five minutes:
现在我想从过去五分钟的数据框中过滤行：
only_last_5_minutes = df.filter(
datetime.strptime(df.ts, '%d-%b-%y %I.%M.%S.%f %p') > datetime.now() - timedelta(minutes=5)
)
However, this does not work and I get this message
但是，这不起作用，我收到此消息
TypeError: strptime() argument 1 must be string, not Column
It looks like I have wrong application of column operation and it seems to me I have to create a lambda function to filter each column that satisfies the desired condition, but being a newbie to Python and lambda expression in particular, I don't know how to create my filter correct. Please advise.
看起来我对列操作的应用有误，在我看来我必须创建一个 lambda 函数来过滤满足所需条件的每一列，但作为 Python 和 lambda 表达式的新手，我不知道如何正确创建我的过滤器。请指教。
P.S. I prefer to express my filters as Python native (or SparkSQL) rather than a filter inside Hive sql query expression 'WHERE'.
PS 我更喜欢将我的过滤器表示为 Python 本机（或 SparkSQL），而不是 Hive sql 查询表达式“WHERE”中的过滤器。
preferred:
首选：
df = sqlContext.sql("SELECT * FROM my_table")
df.filter( // filter here)
not preferred:
不推荐：
df = sqlContext.sql("SELECT * FROM my_table WHERE...")
采纳答案by zero323
It is possible to use user defined function.
可以使用用户定义的函数。
from datetime import datetime, timedelta
from pyspark.sql.types import BooleanType, TimestampType
from pyspark.sql.functions import udf, col
def in_last_5_minutes(now):
def _in_last_5_minutes(then):
then_parsed = datetime.strptime(then, '%d-%b-%y %I.%M.%S.%f %p')
return then_parsed > now - timedelta(minutes=5)
return udf(_in_last_5_minutes, BooleanType())
Using some dummy data:
使用一些虚拟数据：
df = sqlContext.createDataFrame([
(1, '14-Jul-15 11.34.29.000000 AM'),
(2, '14-Jul-15 11.34.27.000000 AM'),
(3, '14-Jul-15 11.32.11.000000 AM'),
(4, '14-Jul-15 11.29.00.000000 AM'),
(5, '14-Jul-15 11.28.29.000000 AM')
], ('id', 'datetime'))
now = datetime(2015, 7, 14, 11, 35)
df.where(in_last_5_minutes(now)(col("datetime"))).show()
And as expected we get only 3 entries:
正如预期的那样，我们只得到 3 个条目：
+--+--------------------+
|id| datetime|
+--+--------------------+
| 1|14-Jul-15 11.34.2...|
| 2|14-Jul-15 11.34.2...|
| 3|14-Jul-15 11.32.1...|
+--+--------------------+
Parsing datetime string all over again is rather inefficient so you may consider storing TimestampTypeinstead.
再次解析日期时间字符串效率很低，因此您可以考虑存储TimestampType。
def parse_dt():
def _parse(dt):
return datetime.strptime(dt, '%d-%b-%y %I.%M.%S.%f %p')
return udf(_parse, TimestampType())
df_with_timestamp = df.withColumn("timestamp", parse_dt()(df.datetime))
def in_last_5_minutes(now):
def _in_last_5_minutes(then):
return then > now - timedelta(minutes=5)
return udf(_in_last_5_minutes, BooleanType())
df_with_timestamp.where(in_last_5_minutes(now)(col("timestamp")))
and result:
结果：
+--+--------------------+--------------------+
|id| datetime| timestamp|
+--+--------------------+--------------------+
| 1|14-Jul-15 11.34.2...|2015-07-14 11:34:...|
| 2|14-Jul-15 11.34.2...|2015-07-14 11:34:...|
| 3|14-Jul-15 11.32.1...|2015-07-14 11:32:...|
+--+--------------------+--------------------+
Finally it is possible to use raw SQL query with timestamps:
最后，可以使用带有时间戳的原始 SQL 查询：
query = """SELECT * FROM df
WHERE unix_timestamp(datetime, 'dd-MMM-yy HH.mm.ss.SSSSSS a') > {0}
""".format(time.mktime((now - timedelta(minutes=5)).timetuple()))
sqlContext.sql(query)
Same as above it would be more efficient to parse date strings once.
与上面相同，解析一次日期字符串会更有效。
If column is already a timestampit possible to use datetimeliterals:
如果列已经是 atimestamp可以使用datetime文字：
from pyspark.sql.functions import lit
df_with_timestamp.where(
df_with_timestamp.timestamp > lit(now - timedelta(minutes=5)))
EDIT
编辑
Since Spark 1.5 you can parse date string as follows:
从 Spark 1.5 开始，您可以按如下方式解析日期字符串：
from pyspark.sql.functions import from_unixtime, unix_timestamp
from pyspark.sql.types import TimestampType
df.select((from_unixtime(unix_timestamp(
df.datetime, "yy-MMM-dd h.mm.ss.SSSSSS aa"
))).cast(TimestampType()).alias("datetime"))
