如何在Spark中从Kafka JSON数据获取各站点最低温度及对应ID？

嘿，我来帮你搞定这个需求！你已经有Kafka生产者在发JSON格式的站点温度数据，现在要用Spark消费者把这些数据转成键值对，再从100个站点里揪出温度最低的那个（或者那些）对应的站点ID对吧？下面一步步来，给你整明白：

首先得确保你的Spark项目里有对应的Kafka连接器依赖，不然连不上Kafka。如果是用Maven的话，把这段加到你的pom.xml里（注意对应Spark和Kafka的版本，别乱配）：

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql-kafka-0-10_2.12</artifactId>
    <version>3.3.0</version> <!-- 换成你的Spark版本 -->
</dependency>

如果是Python用pip的话，直接装对应的pyspark包就行，不过要保证环境里的Spark版本和包匹配。

咱们用Spark Structured Streaming来处理，比旧的Streaming API更灵活好用。首先创建SparkSession，然后从Kafka读数据，提取出JSON格式的value字段，再解析成包含site_id（站点ID）和temperature（温度）的DataFrame——这就相当于转成键值对啦。

Python代码示例： #

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StringType, DoubleType
from pyspark.sql.functions import from_json, col

# 创建SparkSession
spark = SparkSession.builder \
    .appName("KafkaTemperatureMinFinder") \
    .getOrCreate()

# 定义JSON数据的Schema，要和生产者发的结构完全匹配！
# 假设你的JSON是{"site_id": "site_001", "temperature": 18.5}这种格式
temperature_schema = StructType() \
    .add("site_id", StringType()) \
    .add("temperature", DoubleType())

# 从Kafka读取流数据
kafka_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "your-kafka-broker:9092")  # 换成你的Kafka地址
    .option("subscribe", "your-topic-name")  # 换成你要消费的主题
    .load()

# 解析JSON数据，转成结构化的DataFrame
parsed_df = kafka_df.select(
    from_json(col("value").cast(StringType()), temperature_schema).alias("data")
).select("data.*")

# 到这里，parsed_df就是包含site_id和temperature的键值对形式的DataFrame了

如果是批处理（一次性消费所有历史数据），把readStream换成read就行，其他逻辑一样。

接下来就是核心逻辑：找最低温度，然后把对应的站点ID找出来。这里分两种情况：

情况1：全局统计（流处理用Complete模式） #

如果是流处理，要输出全局的最低温，就得用complete输出模式，每次更新都输出当前所有数据里的最低温结果：

# 先算出全局最低温度值
min_temp_df = parsed_df.agg({"temperature": "min"}).withColumnRenamed("min(temperature)", "min_temp")

# 把原数据和最低温度关联，找出所有温度等于最低温的站点
result_df = parsed_df.join(min_temp_df, parsed_df.temperature == min_temp_df.min_temp) \
    .select("site_id", "temperature")

# 输出结果到控制台（生产环境可以换成写文件、数据库等）
query = result_df.writeStream \
    .outputMode("complete") \
    .format("console") \
    .option("checkpointLocation", "/path/to/checkpoint")  # 容错用的checkpoint路径
    .start()

query.awaitTermination()

情况2：批处理一次性计算 #

如果是批处理，逻辑更简单，直接聚合+关联就行：

# 批处理版本
min_temp = parsed_df.agg({"temperature": "min"}).collect()[0][0]
result_df = parsed_df.filter(col("temperature") == min_temp).select("site_id", "temperature")

# 输出结果
result_df.show()

额外注意点： #

• Schema匹配：一定要保证你定义的temperature_schema和生产者发送的JSON结构完全一致，比如字段名、数据类型，不然解析会出null值。
• 空值处理：如果有些站点的温度字段是空的，记得先过滤掉，比如加个.filter(col("temperature").isNotNull())，不然会影响最小值计算。
• 多站点同温：上面的代码会把所有温度等于最低温的站点都列出来，如果你只需要一个，加个.limit(1)就行，但建议保留所有符合的，避免漏数据。
• Scala版本：如果用Scala写，逻辑完全一样，只是语法稍有不同，比如Schema定义用StructType(Seq(StructField("site_id", StringType), ...))，函数调用也类似。

这样一套下来，就能完美实现你的需求啦！

内容的提问来源于stack exchange，提问作者huny