src/services/risk.py
BeforeProposed
+1−1
11from pyspark.sql import functions as F
22
33def risk(frame,times):
4 condition=(F.col('s.group')==F.col('t.group')) & (F.col('s.entry')<=F.col('t.time')) & (F.col('s.time')>F.col('t.time'))
4 condition=(F.col('s.group')==F.col('t.group')) & (F.col('s.entry')<=F.col('t.time')) & (F.col('s.time')>=F.col('t.time'))
55 return times.alias('t').join(frame.alias('s'),condition).groupBy(F.col('t.group').alias('group'),F.col('t.time').alias('time')).agg(F.count('*').alias('risk'))