structedstreaming 如何计算异常用户