Spark 和 Kafka 处理 API 请求与返回数据DEMO

以下是一个更详细的使用 Spark 和 Kafka 处理 API 请求与返回数据，并保障其正常性的示例代码。这个示例代码增加了一些错误处理和数据校验的逻辑：

from pyspark import SparkContext from pyspark.streaming import StreamingContext from pyspark.streaming.kafka import KafkaUtils import json import requests import time  # 创建 SparkContext 和 StreamingContext sc = SparkContext(appName="KafkaStreamingWithAPI") ssc = StreamingContext(sc, 5)  # 每 5 秒处理一次数据  # Kafka 配置 kafkaParams = {     "bootstrap.servers": "localhost:9092",     "group.id": "spark-streaming-group" }  # 定义要订阅的 Kafka 主题 topics = ["your_kafka_topic"]  # 从 Kafka 读取数据 kafkaStream = KafkaUtils.createDirectStream(ssc, topics, kafkaParams)  # API 端点 api_endpoint = "http://your_api_endpoint.com/api"  # 处理接收到的消息 def process_data(rdd):     if not rdd.isEmpty():         data = rdd.map(lambda x: json.loads(x[1]))                  # 数据校验         valid_data = [item for item in data if 'key_you_expect' in item]                  for item in valid_data:             try:                 # 发送 API 请求                 response = requests.post(api_endpoint, json=item)                 if response.status_code == 200:                     # 处理成功响应                     print("API request successful")                 else:                     # 处理错误响应                     print(f"API request failed with status code: {response.status_code}")             except requests.exceptions.RequestException as e:                 # 处理请求异常                 print(f"Request exception: {e}")  # 对 Kafka 数据流进行处理 kafkaStream.foreachRDD(process_data)  # 启动 Spark Streaming 上下文 ssc.start() ssc.awaitTermination() 

在上述代码中：

• 增加了对数据的校验，只处理包含特定键的有效数据。
• 在发送 API 请求时，处理了不同的响应状态码和可能的请求异常。

请注意，您需要根据实际的 API 接口和数据要求来进一步完善数据校验、错误处理和响应处理的逻辑。同时，还需要将 api_endpoint 替换为实际的 API 端点，并根据 API 的要求设置正确的请求方法和数据格式。