当前位置: 首页 > article >正文

【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有几个优势:

  1. 简化过程:简化了ADB 从各种数据源抽取数据到Delta Table的过程。它自动检测特定目录中的新文件
  2. 时效性高:接近实时地,高效地加载到表中。
  3. 可以处理大数据量:同时它还能应对大体量的数据,不像某些数据集成工具,只适合短时、少量的数据处理(因为通常这些工具目标是处理逻辑而不是数据量)。
  4. 易用:不需要写复杂的代码来实现文件发现和数据加载。还能快速适应数据结构的变化。
  5. 支持数据源多:无缝对接Azure生态圈中的数据源, 如Event Hubs,Azure Blob Storage。
  6. 支持预处理:这个大部分的ETL 工具都支持,在这里只是说明它也支持。
  7. 可靠性和一致性:通过事务控制数据的增删改操作。
  8. 支持多种数据格式:包括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等事件触发来监控和加载新的文件。
在这里插入图片描述


http://www.kler.cn/a/589324.html

相关文章:

  • 随笔小记-本人常用桌面应用(流程图-boardmix,截图-snipaste,文件比较-beyond compare,远程控制-向日葵,解压-360压缩)
  • 机试准备第18天
  • Python的类和对象(4)
  • 在Django模型中的Mysql安装
  • oracle 基础知识之 多表查询
  • JVM---Java 类生命周期与类加载机制
  • 电子电气架构 --- 智能电动汽车的品牌竞争转变
  • 【失败了】LazyGraphRAG利用本地ollama提供Embedding model服务和火山引擎的deepseek API构建本地知识库
  • 面试系列|蚂蚁金服技术面【3】
  • C语言内存函数讲解
  • 10-SDRAM控制器的设计—— signaltap 调试
  • iptables与firewall的区别,从不同的角度讲解
  • 基于金融产品深度学习推荐算法详解【附源码】
  • C++类:特殊的数据成员
  • 鸿蒙跳转到系统设置app界面
  • JAVA(8)-数组
  • 07.Python基础4
  • Linux中安装MySQL
  • 我又又又又又又又更新了~~~纯手工编写C++画图,有注释~~~
  • 解决git fetch 成功后还是不能checkout到fetch分支