df = spark.read.csv('/export/20001-battle-0-2019-07-29.log',sep='|')
json
df = spark.read.json('hdfs://bigdata1/export/logging_20190722.log')
df.createOrReplaceTempView("df")
ds2 = spark.sql("select token,min(ctime) as lt,max(ctime) as oft,(max(ctime)-min(ctime)) as online \
from df where ltype='online' group by token").collect()
collect() 直接打印全部结果集
print ds2
[Row(token=u'3', lt=1563851449, oft=1563851979, online=530),
Row(token=u'0B2D0360-AC77-11E9-884A-00163E01F784', lt=1563796293, oft=1563796415, online=122)]
df.select(df.token).filter(df.ltype=='online').filter(df.guid>0).show()
+--------------------+
| token|
+--------------------+
|FFDAC760-AC6F-11E…|
|FFDAC760-AC6F-11E…|
|FFDAC760-AC6F-11E…|
|FFDAC760-AC6F-11E…|
|FFDAC760-AC6F-11E…|
|FFDAC760-AC6F-11E…|
|FFDAC760-AC6F-11E…|
分类