stream根据某个字段分组

要根据某个字段分组，可以使用`groupBy`函数。例如，如果我们有一个包含以下内容的流：

{"name": "Alice", "age": 23}
{"name": "Bob", "age": 25}
{"name": "Charlie", "age": 23}
{"name": "David", "age": 27}

我们可以使用以下代码将其按年龄分组：

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

ssc = StreamingContext(sc, 1)
kafka_stream = KafkaUtils.createStream(ssc, "localhost:2181", "my-group", {"my-topic": 1})
parsed_stream = kafka_stream.map(json.loads)
age_groups = parsed_stream.groupBy(lambda x: x["age"])

现在，`age_groups`是一个键值对的DStream，其中键是年龄，值是该年龄的所有记录。您可以使用`foreachRDD`方法对每个分组进行操作，例如：

def process_group(time, rdd):
    print(f"=== {time} ===")
    for age, records in rdd.collect():
        print(f"Age {age}:")
        for record in records:
            print(f"\t{record}")
    print("")

age_groups.foreachRDD(process_group)

这将输出以下内容：

=== 2022-01-01 00:00:00 ===
Age 23:
	{'name': 'Alice', 'age': 23}
	{'name': 'Charlie', 'age': 23}

Age 25:
	{'name': 'Bob', 'age': 25}

Age 27:
	{'name': 'David', 'age': 27}