如何在AWS Spark作业中用current_date()与hour()读取对应S3分区？

完全可以借助current_date()和hour(current_timestamp())这两个Spark函数来动态定位S3上对应当前日期、小时的分区数据，具体实现分两种常用场景，下面逐一说明：

1. 直接拼接S3路径读取 #

如果你的S3分区结构是类似s3://your-bucket/data/date=2022-09-07/hour=18/这种格式，你可以先通过Spark SQL获取当前日期和小时的字符串值，再拼接成完整的S3路径进行读取。

Scala示例 #

import org.apache.spark.sql.functions._

// 获取格式化后的当前日期字符串（匹配分区的日期格式）
val currentDate = spark.sql("select date_format(current_date(), 'yyyy-MM-dd')").first().getString(0)
// 获取当前小时的字符串值
val currentHour = spark.sql("select cast(hour(current_timestamp()) as string)").first().getString(0)

// 拼接目标S3路径
val targetS3Path = s"s3://your-bucket/your-base-dir/date=$currentDate/hour=$currentHour/"

// 读取分区数据
val df = spark.read.parquet(targetS3Path)

Python示例 #

from pyspark.sql.functions import current_date, hour, current_timestamp

# 获取格式化后的当前日期
current_date_str = spark.sql("select date_format(current_date(), 'yyyy-MM-dd')").collect()[0][0]
# 获取当前小时（转为字符串匹配分区格式）
current_hour_str = spark.sql("select cast(hour(current_timestamp()) as string)").collect()[0][0]

# 拼接S3路径
target_s3_path = f"s3://your-bucket/your-base-dir/date={current_date_str}/hour={current_hour_str}/"

# 读取数据
df = spark.read.parquet(target_s3_path)

2. 分区过滤读取（更高效） #

如果你的数据集是从父路径加载（比如s3://your-bucket/your-base-dir/），或者已经注册为Spark表，直接用函数作为过滤条件筛选分区会更高效——Spark会自动执行分区修剪，只扫描目标分区的数据，避免全路径扫描。

SQL示例 #

SELECT *
FROM parquet.`s3://your-bucket/your-base-dir/`
WHERE date = current_date()
  AND hour = hour(current_timestamp())

DataFrame API示例（Scala） #

val df = spark.read.parquet("s3://your-bucket/your-base-dir/")
  .filter(col("date") === current_date() && col("hour") === hour(current_timestamp()))

关键注意点 #

• 时区一致性：current_date()和current_timestamp()默认使用Spark会话的时区配置，如果你的S3分区使用的是UTC或其他特定时区，需要通过spark.sql.session.timeZone参数统一时区，避免读取到错误的分区。
• 数据类型匹配：如果分区列hour存储的是字符串类型（比如hour="18"），记得把hour(current_timestamp())的结果转为字符串（如cast(hour(current_timestamp()) as string)），否则会因类型不匹配导致过滤失效。

内容的提问来源于stack exchange，提问作者Erdem Çulcu