azure - Azure Blob 存储和 Azure databricks 之间的高效数据检索过程

Question

我正在设计一个新的数据景观，目前正在开发我的概念验证。在这里我使用以下架构:Azure 函数 --> Azure 事件中心 --> Azure Blob 存储 --> Azure 工厂 --> Azure databricks --> Azure SQL 服务器。

我目前正在努力解决的是如何优化“数据检索”以在 Azure Databricks 上提供我的 ETL 流程的想法。

我正在处理事务性工厂数据，这些数据通过前面的 channel 按分钟提交到 Azure Blob 存储。因此，我每天需要处理 86000 个文件。事实上，需要处理大量单独的文件。目前，我使用以下代码来构建当前存在于 azure blob 存储上的文件名列表。接下来，我通过使用循环读取每个文件来检索它们。

我面临的问题是这个过程所花费的时间。当然，我们这里讨论的是需要读取的大量小文件。所以我不期望这个过程在几分钟内完成。

我知道升级 databricks 集群可能会解决问题，但我不确定只有这样才能解决问题，看看在这种情况下我需要传输的文件数量。我正在运行 databricks 的以下代码。

# Define function to list content of mounted folder
def get_dir_content(ls_path):
  dir_paths = ""
  dir_paths = dbutils.fs.ls(ls_path)
  subdir_paths = [get_dir_content(p.path) for p in dir_paths if p.isDir() and p.path != ls_path]
  flat_subdir_paths = [p for subdir in subdir_paths for p in subdir]
  return list(map(lambda p: p.path, dir_paths)) + flat_subdir_paths

filenames = []
paths = 0

mount_point = "PATH"

paths = get_dir_content(mount_point)
for p in paths:
#   print(p)
  filenames.append(p)

avroFile = pd.DataFrame(filenames)
avroFileList = avroFile[(avroFile[0].str.contains('.avro')) & (avroFile[0].str.contains('dbfs:/mnt/PATH'))]

avro_result = []
# avro_file = pd.DataFrame()
avro_complete = pd.DataFrame()
for i in avroFileList[0]:
  avro_file = spark.read.format("avro").load(i)
  avro_result.append(avro_file)

最后，我对所有这些文件进行联合以创建它们的一个数据帧。

# Schema definiëren op basis van 
avro_df = avro_result[0]

# Union all dataframe
for i in avro_result:
  avro_df = avro_df.union(i)

display(avro_df)

我想知道如何优化这个过程。按分钟输出的原因是，一旦我们有了分析报告架构(我们只需要日常流程)，我们计划稍后构建“近实时洞察”。

Answer 1

我建议您查看Azure Databricks Autoloader，而不是列出文件，然后单独阅读它们。反而。它可以使用通知来查找哪些新文件上传到 blob 存储，而不是列出文件。

它还可以在某个时间点处理多个文件，而不是一个接一个地读取它们并进行联合。

如果不需要连续处理数据，那么可以使用.trigger(once=True)来模拟数据的批量加载。