【Azure 架构师学习笔记】- Azure Databricks (22) --Autoloader
本文属于【Azure 架构师学习笔记】系列。
本文属于【Azure Databricks】系列。
接上文 【Azure 架构师学习笔记】- Azure Databricks (21) --费用相关
前言
Databricks家里在Apache Spark之上，是企业级的应对大规模数据处理的通用平台， 可以运行在AWS， Azure和GCP 之上。
作为数据处理平台， ETL 必不可少，虽然在特定平台比如Azure上可以通过如ADF来实现数据抽取，但是这样对于云平台间迁移并没有什么帮助。
Databricks自带了一个Autoloader功能，本文将介绍一下这个工具。
ADB上的Autoloader
Autoloader是Databricks中的一个“机制”，用于从data lake中获取数据。它的强处在于不需要配置一些列的触发器来处理data lake中的新数据，而是如其名“auto”地把新文件推到流处理作业中。
Autoloader有几个优势：
- 简化过程：简化了ADB 从各种数据源抽取数据到Delta Table的过程。它自动检测特定目录中的新文件
- 时效性高：接近实时地，高效地加载到表中。
- 可以处理大数据量：同时它还能应对大体量的数据，不像某些数据集成工具，只适合短时、少量的数据处理（因为通常这些工具目标是处理逻辑而不是数据量）。
- 易用：不需要写复杂的代码来实现文件发现和数据加载。还能快速适应数据结构的变化。
- 支持数据源多：无缝对接Azure生态圈中的数据源， 如Event Hubs，Azure Blob Storage。
- 支持预处理：这个大部分的ETL 工具都支持，在这里只是说明它也支持。
- 可靠性和一致性：通过事务控制数据的增删改操作。
- 支持多种数据格式：包括JSON, CSV, PARQUET, AVRO,ORC,TEXT, BINARY等文件，也广泛支持3种云平台的存储服务。
组件
- Cloud Files：在Databricks中提供大数据集的分布式文件存储，Autoloader通过CloudFiles管理和存储数据文件的进入。
- CloudNotification：通过启用事件驱动工作流来监听云存储上的变更。当新文件出现在制定的目录时，cloudnotification触发Autoloader去初始化数据处理。
演示
接下来演示一下简单的使用，首先我们需要有一个ADB，还要有一个ADLS Gen2。 按照前文的配置 【Azure 架构师学习笔记】- Azure Databricks (14) – 搭建Medallion Architecture part 2的环境来配置。
新建一个文件夹raw_data， 然后 把测试数据放进去。
在notebook中执行下面的代码。
# 无需任何spark.conf配置，直接使用UC
spark.sql("USE CATALOG george_demo")
spark.sql("CREATE SCHEMA IF NOT EXISTS bronze") # 如果Schema不存在
spark.sql("USE SCHEMA bronze")
# Auto Loader配置
adls_path = "abfss://bronze@medallionadls01.dfs.core.windows.net/raw_data/"
schema_path = "abfss://bronze@medallionadls01.dfs.core.windows.net/schemas/"
checkpoint_path = "abfss://bronze@medallionadls01.dfs.core.windows.net/system/checkpoints/"
# 流式读取（事件驱动模式）
bronze_stream = (spark.readStream
.format("cloudFiles")
.option("cloudFiles.format", "parquet")
.option("cloudFiles.schemaLocation", schema_path)
.option("cloudFiles.useNotifications", "false")
.load(adls_path)
)
# 写入UC托管表（三级命名空间）
target_table = "george_demo.bronze.transactions" # Catalog.Schema.Table
(bronze_stream.writeStream
.format("delta")
.option("mergeSchema", "true")
.trigger(availableNow=True)
.option("checkpointLocation", checkpoint_path)
.toTable(target_table)
)
执行前的截图：
执行代码：
执行后出现了新的表transactions
这是一个简单的演示，后续可以通过下图的schedule，或者借助event hub等事件触发来监控和加载新的文件。
