使用 Spark 获取天气数据：完整指南和代码示例

以下是一个使用 Spark 获取天气数据的模块的示例，它演示了如何使用 Spark 从公开的 API 获取数据，然后解析和显示数据。

导入必要的库和模块

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

创建 SparkSession 对象

spark = SparkSession.builder.appName('WeatherData').getOrCreate()

定义天气数据的结构

weather_schema = StructType([
    StructField('name', StringType(), True),
    StructField('coord', StructType([
        StructField('lon', DoubleType(), True),
        StructField('lat', DoubleType(), True)
    ]), True),
    StructField('weather', StringType(), True),
    StructField('main', StructType([
        StructField('temp', DoubleType(), True),
        StructField('pressure', DoubleType(), True),
        StructField('humidity', DoubleType(), True)
    ]), True),
    StructField('wind', StructType([
        StructField('speed', DoubleType(), True),
        StructField('deg', DoubleType(), True)
    ]), True),
    StructField('clouds', StructType([
        StructField('all', DoubleType(), True)
    ]), True),
    StructField('dt', DoubleType(), True),
    StructField('sys', StructType([
        StructField('country', StringType(), True),
        StructField('sunrise', DoubleType(), True),
        StructField('sunset', DoubleType(), True)
    ]), True),
    StructField('timezone', DoubleType(), True),
    StructField('id', DoubleType(), True),
    StructField('cod', DoubleType(), True)
])

从 API 获取天气数据

url = 'http://api.openweathermap.org/data/2.5/weather?q=London&appid={your_api_key}'
weather_data = spark.read.json(url, schema=weather_schema)

注意： 将 {your_api_key} 替换为您的 OpenWeatherMap API 密钥。

展示数据

weather_data.show()

完整代码：

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

spark = SparkSession.builder.appName('WeatherData').getOrCreate()

weather_schema = StructType([
    StructField('name', StringType(), True),
    StructField('coord', StructType([
        StructField('lon', DoubleType(), True),
        StructField('lat', DoubleType(), True)
    ]), True),
    StructField('weather', StringType(), True),
    StructField('main', StructType([
        StructField('temp', DoubleType(), True),
        StructField('pressure', DoubleType(), True),
        StructField('humidity', DoubleType(), True)
    ]), True),
    StructField('wind', StructType([
        StructField('speed', DoubleType(), True),
        StructField('deg', DoubleType(), True)
    ]), True),
    StructField('clouds', StructType([
        StructField('all', DoubleType(), True)
    ]), True),
    StructField('dt', DoubleType(), True),
    StructField('sys', StructType([
        StructField('country', StringType(), True),
        StructField('sunrise', DoubleType(), True),
        StructField('sunset', DoubleType(), True)
    ]), True),
    StructField('timezone', DoubleType(), True),
    StructField('id', DoubleType(), True),
    StructField('cod', DoubleType(), True)
])

url = 'http://api.openweathermap.org/data/2.5/weather?q=London&appid={your_api_key}'
weather_data = spark.read.json(url, schema=weather_schema)

weather_data.show()

这个示例演示了如何使用 Spark 从公开的 API 获取天气数据，然后使用结构化数据类型进行解析和显示。您可以根据需要修改代码以获取其他城市的天气数据或处理更多数据。