Spark Streaming 实战：计算商品单价乘以总价

下面是一个简单的示例代码，用于实现计算商品的单价乘以商品的总价的功能。

from pyspark import SparkContext
from pyspark.streaming import StreamingContext

# 创建SparkContext和StreamingContext
sc = SparkContext('local[2]', 'SparkStreamingExample')
ssc = StreamingContext(sc, 1)

# 创建一个DStream，从socket接收数据流
lines = ssc.socketTextStream('localhost', 9999)

# 将每行数据按空格分割为单词，并转换为(key, value)对
words = lines.flatMap(lambda line: line.split(' '))

# 过滤出包含商品名称和单价的记录，并将其转换为(key, value)对
product_prices = words.filter(lambda word: word.startswith('product_')).map(lambda word: (word, float(word.split('_')[1])))

# 计算商品的总价，并将其转换为(key, value)对
total_prices = product_prices.reduceByKey(lambda x, y: x + y)

# 计算商品的单价乘以总价，并打印结果
result = total_prices.map(lambda x: (x[0], x[1] * x[1]))
result.pprint()

# 启动StreamingContext
ssc.start()
ssc.awaitTermination()

要运行此代码，需要确保在本地主机的端口9999上运行一个socket服务器，以接收输入数据流。可以使用以下命令在终端上运行一个简单的socket服务器：

nc -lk 9999

然后，您可以通过在终端上输入商品名称和单价来向服务器发送数据，例如：

product_1 10
product_2 20

接下来，您将在终端上看到计算结果，例如：

('product_1', 100.0)
('product_2', 400.0)

请注意，此示例代码假定输入的每行只包含一个商品名称和一个商品单价，以空格分隔。根据实际情况，您可能需要根据输入数据的格式进行适当的修改。