kafka整合streaming完整代码

这里提供一个简单的Kafka和Spark Streaming整合的完整代码，用于从Kafka主题中读取数据并进行处理：

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

# 设置Kafka相关配置
kafka_broker = "localhost:9092"
kafka_input_topic = "input"
kafka_output_topic = "output"

# 设置Spark Streaming的批处理时间为2秒
ssc = StreamingContext(sparkContext, 2)

# 从Kafka主题中读取数据
kafka_stream = KafkaUtils.createDirectStream(ssc, [kafka_input_topic], {"metadata.broker.list": kafka_broker})

# 处理接收到的数据
def process_data(data):
    # 对数据进行处理，这里只是简单的将数据转换为大写
    processed_data = data.upper()
    return processed_data

# 将处理后的数据写入Kafka主题
def write_to_kafka(partition):
    producer = KafkaProducer(bootstrap_servers=[kafka_broker])
    for record in partition:
        producer.send(kafka_output_topic, json.dumps(record).encode('utf-8'))

# 对接收到的数据进行处理并将处理后的数据写入Kafka主题
processed_data = kafka_stream.map(lambda x: process_data(x[1]))
processed_data.foreachRDD(lambda rdd: rdd.foreachPartition(write_to_kafka))

# 启动Spark Streaming应用
ssc.start()
ssc.awaitTermination()

这个代码片段基于Python编写，使用了Spark Streaming和Kafka的Python客户端库。在代码中，我们首先定义了Kafka主题和Spark Streaming的配置。然后，我们创建了一个Spark Streaming上下文，并使用KafkaUtils从Kafka主题中读取数据。

接下来，我们定义了两个函数来处理接收到的数据。process_data函数将数据转换为大写。write_to_kafka函数将处理后的数据写入另一个Kafka主题中。

最后，我们使用Spark Streaming的map和foreachRDD函数来对接收到的数据进行处理，并将处理后的数据写入Kafka主题中。我们通过启动应用程序并等待它的终止来启动应用程序。

请注意，这个代码片段只是一个简单的演示，并没有考虑到错误处理和性能优化。在实际应用中，您需要更全面地考虑这些问题。