Spark.read() 一次多条路径，而不是在 for 循环中一个接一个

Posted 2023-04-15

【中文标题】Spark.read() 一次多条路径，而不是在 for 循环中一个接一个

【英文标题】：Spark.read() multiple paths at once instead of one-by-one in a for loop

【发布时间】：2020-05-16 07:24:55

【问题描述】：

我正在运行以下代码：

list_of_paths 是一个包含以 avro 文件结尾的路径的列表。例如，

['folder_1/folder_2/0/2020/05/15/10/41/08.avro', 'folder_1/folder_2/0/2020/05/15/11/41/08.avro', 'folder_1/folder_2/0/2020/05/15/12/41/08.avro']

注意：以上路径存储在 Azure Data Lake 存储中，以下过程在 Databricks 中执行

spark.conf.set("fs.azure.account.key.0.dfs.core.windows.net".format(storage_account_name), storage_account_key)
spark.conf.set("spark.sql.execution.arrow.enabled", "false")
begin_time = time.time()

for i in range(len(list_of_paths)):

    try:
      read_avro_data,avro_decoded=None,None

      #Read paths from Azure Data Lake "abfss"
      read_avro_data=spark.read.format("avro").load("abfss://[e-mail]/2".format(storage_container_name, storage_account_name, list_of_paths[i]))

    except Exception as e:
      custom_log(e)

架构

read_avro_data.printSchema()

root
 |-- SequenceNumber: long (nullable = true)
 |-- Offset: string (nullable = true)
 |-- EnqueuedTimeUtc: string (nullable = true)
 |-- SystemProperties: map (nullable = true)
 |    |-- key: string
 |    |-- value: struct (valueContainsNull = true)
 |    |    |-- member0: long (nullable = true)
 |    |    |-- member1: double (nullable = true)
 |    |    |-- member2: string (nullable = true)
 |    |    |-- member3: binary (nullable = true)
 |-- Properties: map (nullable = true)
 |    |-- key: string
 |    |-- value: struct (valueContainsNull = true)
 |    |    |-- member0: long (nullable = true)
 |    |    |-- member1: double (nullable = true)
 |    |    |-- member2: string (nullable = true)
 |    |    |-- member3: binary (nullable = true)
 |-- Body: binary (nullable = true) 
# this is the content of the AVRO file.

行数和列数

print("ROWS: ", read_avro_data.count(), ", NUMBER OF COLUMNS: ", len(read_avro_data.columns))

ROWS:  2 , NUMBER OF COLUMNS:  6

我想要不每次迭代读取 1 个 AVRO 文件，因此一次迭代读取 2 行内容。相反，我想一次读取所有 AVRO 文件。所以我最终的 spark DataFrame 中有 2x3 = 6 行内容。

这对 spark.read() 可行吗？类似于以下内容：

spark.read.format("avro").load("abfss://[e-mail]/folder_1/folder_2/0/2020/05/15/*")

[更新] 对通配符（*）的误解深表歉意。这意味着所有 AVRO 文件都在同一个文件夹中。但是，每个 AVRO 文件我有 1 个文件夹。所以 3 个 AVRO 文件，3 个文件夹。在这种情况下，通配符将不起作用。下面回答的解决方案是使用带有路径名的列表 []。

提前感谢您的帮助和建议。

【参考方案1】：

load(path=None, format=None, schema=None, **options) 此方法将接受单个路径或路径列表。

例如，您可以直接传递路径列表，如下所示

spark.read.format("avro").load(["/tmp/dataa/userdata1.avro","/tmp/dataa/userdata2.avro"]).count()

1998

【讨论】：

您可以使用通配符 '*' 之类的东西，它会自动并行读取所有 avro 文件。性能不会有问题spark.read.format('avro').load('python/test_support/sql/*')

最好将所有路径传递给 spark，它将并行加载文件。如果你使用 foreach ，它会按顺序加载文件。

我已经测试并更新了答案。如果对解决问题有帮助，请采纳或点赞？

@NikSp 你为什么不继续提供通配符。您根本不需要阅读所有路径。由于您已经连接到 ADLS，您可以简单地硬编码到根文件夹并传递 *.. 根路径为 abfss://[e-mail]/rootpath/*。

@RohithRangaraju 是的，实际上我尝试使用通配符 (*) 但由于每个 AVRO 文件有 1 个文件夹，并且我想要来自所有可能不同文件夹的信息（按年/月/日/小时/分钟），我只能使用通配符如果我对月、日、小时、分钟的可能组合进行循环。由于我想避免 for 循环并优化 AVRO 文件的读取，这就是我使用多路径列表方式的原因:)。