在 Azure Databricks 中的日期范围之间读取镶木地板文件的有效方法

Posted 2023-04-17

【中文标题】在 Azure Databricks 中的日期范围之间读取镶木地板文件的有效方法

【英文标题】：Efficient way of reading parquet files between a date range in Azure Databricks

【发布时间】：2019-02-28 16:39:07

【问题描述】：

我想知道下面的伪代码是否是从 PySpark（Azure Databricks）读取存储在 Azure Data Lake 中的日期范围之间的多个镶木地板文件的有效方法。注意：parquet 文件不按日期分区。

我使用 uat/EntityName/2019/01/01/EntityName_2019_01_01_HHMMSS.parquet 约定将数据存储在 ADL 中，如 Nathan Marz 的《大数据》一书中所建议的，稍作修改（使用 2019 而不是 year=2019）。

使用 * 通配符读取所有数据：

df = spark.read.parquet(uat/EntityName/*/*/*/*)

添加一个列 FileTimestamp，使用字符串操作从 EntityName_2019_01_01_HHMMSS.parquet 中提取时间戳并转换为 TimestampType()

df.withColumn(add timestamp column)

使用过滤器获取相关数据：

start_date = '2018-12-15 00:00:00'
end_date = '2019-02-15 00:00:00'
df.filter(df.FileTimestamp >= start_date).filter(df.FileTimestamp < end_date)

基本上我使用 PySpark 来模拟 U-SQL 中可用的简洁语法：

@rs = 
  EXTRACT 
      user    string,
      id      string,
      __date  DateTime
  FROM 
    "/input/data-__date:yyyy-__date:MM-__date:dd.csv"
  USING Extractors.Csv();

@rs = 
  SELECT * 
  FROM @rs
  WHERE 
    date >= System.DateTime.Parse("2016/1/1") AND
    date < System.DateTime.Parse("2016/2/1");

【参考方案1】：

对数据进行分区的正确方法是对数据使用格式 year=2019、month=01 等。

当您使用以下过滤器查询此数据时：

df.filter(df.year >= myYear)

然后 Spark 将只读取相关文件夹。

过滤列名称准确地出现在文件夹名称中是非常重要的。请注意，当您使用 Spark（例如按年、月、日）写入分区数据时，它不会将分区列写入 parquet 文件。相反，它们是从路径中推断出来的。这确实意味着您的数据框在编写时将需要它们。当您从分区源读取时，它们也将作为列返回。

如果您无法更改文件夹结构，您始终可以手动减少文件夹以供 Spark 使用正则表达式或 Glob 读取 - 本文应提供更多上下文 Spark SQL queries on partitioned data using Date Ranges。但显然这更加手动和复杂。

更新：进一步的例子Can I read multiple files into a Spark Dataframe from S3, passing over nonexistent ones?

也来自“Spark - 权威指南：大数据处理变得简单” 比尔·钱伯斯：

分区是一种工具，可让您控制存储的数据 （以及在哪里）在你写它的时候。当您将文件写入分区时 目录（或表），您基本上将列编码为文件夹。什么 这允许您在读取时跳过大量数据 稍后，允许您仅读取与您的问题相关的数据 而不必扫描完整的数据集。 ...

这可能是您可以在以下情况下使用的最简单的优化 你有一个读者经常过滤的表格 操纵。例如，日期对于 分区，因为在下游，我们通常只想查看 前一周的数据（而不是扫描整个记录列表）。

【讨论】：

嗨，西蒙，感谢您的回复。我无法从路径中理解列推理机制。我一直在寻找更权威的文档。任何指针？查询模式将是一个日期范围，例如检索当前月份或最近 3 个月的数据。我同意球是不可行的。采用隐式列 year=2019 使我与 glob 处于同一位置。您是否认为将 Timestamp 列与 year=2019 文件夹结构一起添加最能实现日期范围过滤和高效的数据检索。我在哪里查看所涉及的累积读取数/IOP？

我添加了一个更新来尝试扩展它。另请注意，在 SparkUI 中，您可以查看输入数据集的大小，以了解读取了多少数据以及过滤应用的预输入。

让我检查并恢复

嗨，西蒙，您提供的解决方案奏效了。我正在使用 ADF 构建不正确的文件结构的年/月/日格式。相反，我求助于 Databricks 来构建文件夹结构，并在写入时指定了分区键。这创建了所需的文件夹结构，当我合并文件时，它在合并的 df 中给出了隐式列年、月、日。感谢您的帮助。