如何在Databricks中通过指定时间范围从Kinesis/Kafka读取历史数据？

不能直接通过Kinesis数据源的配置同时设置起始和结束位置，原因如下：

• Spark的Kinesis流式数据源（spark.readStream）是为持续监听、处理新数据设计的，本身没有提供直接配置结束位置的参数，initialPosition仅用来指定消费的起始点。
• 要读取2023年3月8日至3月14日期间到达的数据，需要通过「起始位置配置+数据过滤+流停止控制」的组合方式实现。

具体实现步骤 #

1. 设置起始位置：保留你代码中initialPosition的配置，指定从2023年3月8日0点PDT开始消费。
2. 过滤时间范围数据：利用Kinesis记录自带的approximateArrivalTimestamp字段（数据到达Kinesis的时间戳），过滤出3月14日23:59:59 PDT之前的记录。
3. 控制流停止：当符合时间范围的数据全部处理完成后，手动停止流查询，或者通过监控数据输出状态自动停止。

修改后的代码示例 #

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

// 转换时区后定义时间范围（根据PDT实际时区调整）
val startTime = Timestamp.valueOf("2023-03-08 00:00:00")
val endTime = Timestamp.valueOf("2023-03-14 23:59:59")

// 读取Kinesis流
val kinesisStreamDF = spark.readStream
  .format("kinesis")
  .option("streamName", kinesisStreamName)
  .option("region", kinesisRegion)
  .option("initialPosition", """{"at_timestamp": "03/08/2023 00:00:00 PDT", "format": "MM/dd/yyyy HH:mm:ss ZZZ"}""")
  .option("awsAccessKey", awsAccessKeyId)
  .option("awsSecretKey", awsSecretKey)
  .load()

// 过滤指定时间范围内的到达数据
val filteredDataDF = kinesisStreamDF
  .filter(col("approximateArrivalTimestamp").between(startTime, endTime))

// 输出处理结果（示例为Parquet格式）
val streamQuery = filteredDataDF.writeStream
  .outputMode("append")
  .format("parquet")
  .option("path", "/your/output/path")
  .option("checkpointLocation", "/your/checkpoint/path")
  .start()

// 等待流处理完成后手动停止，或根据业务逻辑自动终止
streamQuery.awaitTermination()

补充说明 #

如果你的场景是一次性读取指定时间范围的数据而非持续流式处理，也可以考虑用AWS SDK直接调用Kinesis批处理API读取，或者用Spark批处理模式读取Kinesis数据（通过配置iteratorType等参数），这种方式更适合一次性数据导出的需求。

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