使用PySpark计算每个窗口的用户数

我正在使用Kafka流式传输JSON文件，将每一行作为消息发送。其中一个关键是用户email。

然后我使用PySpark计算每个窗口的唯一用户数，使用他们的电子邮件来识别它们。命令

def print_users_count(count):

print 'The number of unique users is:', count

print_users_count((lambda message: message['email']).distinct().count())
给我下面的错误。我怎样才能解决这个问题？

AttributeError Traceback (most recent call last)
in ()

  2     print 'The number of unique users is:', count
  3 

----> 4 print_users_count((lambda message: message['email']).distinct().count())

AttributeError: 'function' object has no attribute 'distinct'
这是我的PySpark代码：

from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils
import json

try:

sc.stop()

except:

pass  

sc = SparkContext(appName="KafkaStreaming")
sc.setLogLevel("WARN")

ssc = StreamingContext(sc, 60)

Define the PySpark consumer.
kafkaStream = KafkaUtils.createStream(ssc, bootstrap_servers, 'spark-streaming2', {topicName:1})

Parse the incoming data as JSON.
parsed = kafkaStream.map(lambda v: json.loads(v[1]))

Count the number of messages per batch.
parsed.count().map(lambda x:'Messages in this batch: %s' % x).pprint()