青岛网站建设方案策划,wordpress 知更鸟5.2,如何制作简单网页,南宁seo外包服务一、说明 我们探索数据工程中使用的设计模式 - 软件设计中常见问题的可重用解决方案。 以下文章是有关 Python 数据工程系列文章的一部分#xff0c;旨在帮助数据工程师、数据科学家、数据分析师、机器学习工程师或其他刚接触 Python 的人掌握基础知识。 迄今为止#xff0c;… 一、说明 我们探索数据工程中使用的设计模式 - 软件设计中常见问题的可重用解决方案。 以下文章是有关 Python 数据工程系列文章的一部分旨在帮助数据工程师、数据科学家、数据分析师、机器学习工程师或其他刚接触 Python 的人掌握基础知识。 迄今为止本初学者指南包括 第 1 部分Python 包探讨了 Python 模块、Python 包的基础知识以及如何将模块导入您自己的项目。【Python 程序设计】包和数据人员入门【01/8】 第 2 部分介绍了依赖项管理和虚拟环境。【Python 程序设计】数据人员入门【02/8】第 3 部分 项目的最佳实践涵盖了构建项目的 9 个最佳实践和示例。【Python程序设计】 项目的最佳实践【03/8】第 4 部分 Dagster 管道我们探讨了设置 Dagster 项目以及数据资产的关键概念。【Python程序设计】 从 Python 项目到 Dagster Pipelines【04/8】第 5 部分我们将介绍环境变量的重要性以及如何使用它们。【Python程序设计】Python 中的环境变量【05/8】第 6 部分类型提示或类型提示如何减少错误。第7 部分模式或学习设计模式它们是软件设计中常见问题的可重用解决方案。【Python程序设计】 工厂模式【07/8】 二、Python 编程中的工厂模式 在本系列中你了解了数据工程中的 Python 最佳实践以及如何构建更健壮和可扩展的软件。今天我们将采用一种称为设计模式的更高级编程概念它是软件设计中常见问题的可重用解决方案。您可能会在数据工程项目中遇到这些问题。 我们将看看什么是设计模式为什么要使用它们并特别深入探讨一种模式工厂模式。我们还将讨论为什么数据工程师会从使用工厂模式中受益。 工厂模式优雅地简化了数据连接器的创建并使现有数据基础结构更具可扩展性。这对于处理具有不同处理需求的各种数据源的数据工程师特别有用。 工厂模式定义在软件设计中“工厂模式”允许您创建一个类其子类决定要实例化哪个类。工厂模式允许在不指定将要创建的确切对象类的情况下创建对象从而促进代码中的松散耦合和可伸缩性。此模式通常使用工厂方法或工厂类来实现以基于输入条件或条件生成对象的实例。 工厂模式称为“创建设计模式”本文稍后将对此进行说明。 在本课程的这一部分中我们将了解工厂模式如何在数据工程中实现可重用性、可伸缩性和可维护性。 三、 设计模式 101 Python 中的设计模式作为模板工作可以应用于重复性任务或问题因此在数据工程中非常有用。对于 Python 数据工程师来说设计模式为数据处理和集成任务中反复出现的挑战提供了结构化且高效的解决方案。它们还提供共享词汇表促进团队成员之间更清晰的沟通从而实现更一致和协作的软件设计。 3.1 设计模式的类型 Python 中的设计模式以及一般编程中的设计模式通常被认为是中级到高级概念因为它们通常需要了解编程原理、面向对象设计以及识别和抽象更大、更复杂的系统即代码架构中反复出现的问题的能力。Python的设计模式通常分为三种类型 创建模式这些模式包括工厂模式可用于实例化和管理数据库连接确保无论是连接到 SQL 数据库还是 NoSQL 存储该过程都得到简化且一致结构模式提供有关组织和链接不同组件的指导在集成不同的数据源时可能非常宝贵。例如您可以使用适配器模式将来自旧系统的数据与现代分析平台协调从而确保无缝数据流行为模式提供对象之间有效沟通和交互的策略。例如可以使用观察者模式来监视数据集中的更改想象一下每当新数据到达时数据引入过程都会通知多个下游处理任务。 3.2 为什么要在数据工程中使用设计模式 设计模式为数据工程任务提供了明显的优势这些任务与软件设计的核心原则相呼应 声明通过设计模式数据工程师可以定义最终目标或他们想要实现的目标而不会陷入程序应该如何执行的细节中。通过定义你想要什么底层逻辑会处理“如何”。这种抽象简化了过程。可 重用 将设计模式视为蓝图。正如建筑计划可用于建造各种建筑物一样这些模式可以应用于不同的管道或项目确保您的工作不仅限于一个解决方案而是可以多次利用。一致 数据世界是广阔而多样的。通过采用设计模式数据工程师可确保数据资产无论其来源或应用程序如何都遵循一致的结构和行为。这使得数据操作可预测并减少异常从而产生更强大的数据系统。 通过将资产工厂等设计模式集成到数据工程工作流从数据提取、分析数据、数据转换等中我们为更顺畅的操作、更少的错误和更高效的系统铺平了道路确保管理和优化数据以获得最佳结果。 每个设计模式都用于防止在项目缩放时出现特定问题。今天我们将重点介绍工厂模式该模式用于构建多个类似的东西以促进集中配置、标准化测试并在遵守一致性的同时允许灵活性。 四、工厂模式如何工作 工厂模式被归类为创建模式因为它们在 Python 编程中创建对象。它们根据某些条件语句或参数返回不同的对象。 4.1 将对象创建与主应用程序分离 将工厂模式视为公司中的一个专业部门只专注于生产某些产品。该部门负责制造的所有细节公司的其他部门只需在需要时要求产品而不必担心其制造方式。 同样工厂模式负责创建特定对象的所有细节。应用程序的其余部分不需要知道这些对象的创建方式或它们需要哪些参数。它只是要求“工厂”生产对象并信任它来处理其余的。这种分离使代码更清晰、更易于理解。 4.2 Python 中的工厂模式 在 Python 中实现工厂模式特别精简这要归功于它的动态类型和一流的函数。您可以从工厂函数返回不同的类甚至函数而无需太多样板。 此外许多 Python 库和框架利用工厂模式或类似工厂的模式即使它不是显式的或完全相同的。例如像SQLAlchemy这样的ORM对象关系映射库使用工厂来创建数据库会话对象。SQLAlchemy 可以比作工厂模式因为它生成新的会话实例充当数据库通信的主要接口。sessionmaker() 4.3 使用 Python 在数据工程中使用工厂模式示例 Python 的内置功能如装饰器可用于增强工厂模式。例如装饰器可用于向工厂注册类从而扩展工厂的功能而无需显式修改它。 想象一下数据工程中的一个常见场景用于操作来自不同文件格式的数据的数据管道CSV、JSON 和 XML 文件。根据文件类型应应用不同的分析步骤。 我们将使用一个简单的字典作为这些文件解析器的“注册表”并将函数作为我们的工厂。 首先我们将定义解析函数 import csv
import json
import xml.etree.ElementTree as ETdef parse_csv(file_path):with open(file_path, moder) as file:reader csv.reader(file)return list(reader)def parse_json(file_path):with open(file_path, moder) as file:return json.load(file)def parse_xml(file_path):tree ET.parse(file_path)root tree.getroot()return root # youd typically add more logic to process the XML tree 然后我们将定义一个装饰器来注册这些解析器 PARSERS {}def register_parser(file_type):def decorator(fn):PARSERS[file_type] fnreturn fnreturn decorator 我们将注册我们的解析器 register_parser(csv)
def csv_parser(file_path):return parse_csv(file_path)register_parser(json)
def json_parser(file_path):return parse_json(file_path)register_parser(xml)
def xml_parser(file_path):return parse_xml(file_path) 最后我们将编写一个函数来获取正确的解析器并使用工厂来解析文件 def get_parser(file_type):return PARSERS.get(file_type)data_csv get_parser(csv)(data.csv)
data_json get_parser(json)(data.json)
data_xml get_parser(xml)(data.xml) 对于数据工程师来说处理多种文件格式是很常见的并且能够使用新的解析器如XMLParquet等轻松扩展系统至关重要。通过此设置数据工程师只需定义一个新的解析函数并将其注册到装饰器即可轻松扩展系统以支持新的文件类型。 无需接触现有的工厂逻辑使其易于维护和扩展。通过将工厂模式与装饰器一起使用我们可以简化此过程并维护更干净、更模块化的代码。 五、使用工厂模式进行数据工程 数据工程师或数据科学家经常将工厂模式用于日常任务例如批处理、构建实时数据流和 ETL 管道。 例如假设工作流中有各种类型的数据连接例如数据库、文件或 API。您可以使用工厂模式根据当时的需求为您创建正确的连接而不是手动创建与每个连接的连接。可以把它想象成一条装配线在需要时准确地生产你需要的东西而不会用不必要的细节弄乱其余的代码。 5.1 在数据管道中 让我们考虑一个需要连接到不同类型的数据库的场景如MySQL和PostgreSQL。工厂模式可用于根据给定输入创建适当的数据库连接。下面是一个说明此模式的简单示例 让我们首先定义每个数据库的连接 import mysql.connector
import psycopg2def connect_mysql(host, user, password, database):connection mysql.connector.connect(hosthost,useruser,passwordpassword,databasedatabase)return connectiondef connect_postgresql(host, user, password, database):connection psycopg2.connect(hosthost,useruser,passwordpassword,dbnamedatabase)return connection 接下来我们将定义一个装饰器来注册数据库连接 DB_CONNECTIONS {}def register_db_connector(db_type):def decorator(fn):DB_CONNECTIONS[db_type] fnreturn fnreturn decorator 然后我们将注册连接 register_db_connector(mysql)
def mysql_connector(host, user, password, database):return connect_mysql(host, user, password, database)register_db_connector(postgresql)
def postgresql_connector(host, user, password, database):return connect_postgresql(host, user, password, database) 最后我们将编写一个函数来获取正确的连接器并使用工厂来获取适当的数据库连接 def get_db_connector(db_type):if db_type not in DB_CONNECTIONS:raise ValueError(fUnsupported database type: {db_type})return DB_CONNECTIONS[db_type]# Example usage:
mysql_conn get_db_connector(mysql)(localhost, user, password, mydb)
postgres_conn get_db_connector(postgresql)(localhost, user, password, mydb) 通过此设置将来添加对新型数据库连接的支持非常简单。我们首先定义连接函数然后使用装饰器注册它。无需更改其他部件展示了工厂模式的可维护性和可扩展性优势。 六、实际方案中的工厂模式 工厂在以下情况下特别有用 处理可能随时间变化的外部库或系统让您隔离这些更改实现系统的插件或扩展需要出于对象池、延迟初始化或日志记录等目的控制对象实例化 我们将看两个示例说明工厂资产在现实世界中的工作方式。 6.1 示例 1抓取维基百科 当您需要抓取不同类型的页面但希望为所有页面保持一致的界面时工厂模式在网页抓取中特别有用。让我们看看工厂模式如何用于按人口抓取维基百科的国家和依赖关系表 首先请确保安装以下内容 pip install requests
pip install beautifulsoup4 然后定义函数以从维基百科中抓取不同的表。假设维基百科可能有多个表以不同的格式表示这些数据。一个表可能是标准表而另一个表可能是针对移动设备优化的。 from bs4 import BeautifulSoup
import requestsdef scrape_standard_table(url):page requests.get(url)soup BeautifulSoup(page.content, html.parser)# Assuming the first table on the page is the one of interesttable soup.find_all(table)[0]rows table.find_all(tr)data []for row in rows[1:]: # skipping the header rowcolumns row.find_all(td)country columns[0].get_text(stripTrue)population columns[1].get_text(stripTrue)data.append((country, population))return datadef scrape_mobile_table(url):page requests.get(url)soup BeautifulSoup(page.content, html.parser)# Mobile tables might be different, for the sake of example lets assume theyre div-basedtable_div soup.find(div, {class: mobile-table})rows table_div.find_all(div, {class: row})data []for row in rows:country row.find(div, {class: country}).get_text(stripTrue)population row.find(div, {class: population}).get_text(stripTrue)data.append((country, population))return data 然后我们将定义一个装饰器来注册抓取函数 SCRAPERS {}def register_scraper(scraper_type):def decorator(fn):SCRAPERS[scraper_type] fnreturn fnreturn decorator 现在我们将注册我们的抓取函数 register_scraper(standard)
def standard_scraper(url):return scrape_standard_table(url)register_scraper(mobile)
def mobile_scraper(url):return scrape_mobile_table(url) 最后我们将编写我们的函数来获取正确的抓取工具并使用工厂从维基百科获取数据 def get_scraper(scraper_type):if scraper_type not in SCRAPERS:raise ValueError(fUnsupported scraper type: {scraper_type})return SCRAPERS[scraper_type]# Example usage:
url https://en.wikipedia.org/wiki/List_of_countries_and_dependencies_by_population
data_standard get_scraper(standard)(url)
# data_mobile get_scraper(mobile)(url) # if you had a mobile URL 在此示例中添加对将来抓取不同格式的支持如维基百科中的不同表结构很简单定义抓取函数然后使用装饰器注册它。这可确保抓取代码保持模块化且易于扩展而无需修改现有逻辑。 6.2 示例 2数据编排 Dagster 是一个数据编排器可为数据处理的不同阶段从引入到机器学习提供单一管理平台。Dagster帮助安排和观察广泛的数据工程工具Python作为其编程语言。它被数据工程师和数据科学家广泛用于数据科学、数据分析、大数据、机器学习等一系列应用。 Dagster 的功能之一是能够管理资产这些资产是数据计算的输出。资产表示一段数据或具有价值且值得跟踪的计算结果。这可以是数据库中的表、磁盘上的文件、模型工件等。资产通常是管道的输出。 资源工厂是 Dagster 中的一项功能允许用户以声明方式定义资产的生成方式。它们可以被视为通过定义生成资产所需的输入、输出和计算来创建资产的模板。 让我们演练一个简单的示例在该示例中我们将通过在 Dagster 中应用 Factory 模式来重构现有代码块以生成一组资产。 在开始之前请确保安装Dagster pip install dagster dagster-webserver 假设我们是非营利组织数据工程团队的一员。我们有一些现有的代码来查询捐赠者平台的 API 并将结果写入文件CSV 或 JSON。它目前看起来像这样 from dagster import asset
import requests
import csvasset
def volunteers():result requests.get(www.donorplatform.org/api/v1/volunteers)with open(volunteers.csv, w) as f:writer csv.writer(f)writer.writerows(result)asset
def donations():result requests.get(www.donorplatform.org/api/v2/donations)with open(donations.csv, w) as f:writer csv.writer(f)writer.writerows(result)asset
def donors():result requests.get(www.donorplatform.org/api/v1/donors)with open(donors.json, w) as f:f.write(result) 我们的运营团队最近扩大了对捐赠者平台的使用并要求我们从 50 个新的 API 端点运行数据提取。这变得不守规矩且难以管理因为您知道这将需要很长时间数据工程师将偏离他们检索数据的方式并且很难测试。 使用我们新的 python 技能让我们为此应用工厂模式来解决这些问题。 首先让我们定义可以配置的内容。需要为每个资产自定义三个部分 资产的名称它查询的终结点结果保存为的文件类型 因此我们将定义一个 JSON 对象来集中资产的可能配置。 specs [{name: volunteers,endpoint: v1/volunteers,file_type: csv},{name: donations,endpoint: v2/donations,file_type: csv},{name: donors,endpoint: v1/donors,file_type: json}
] 然后我们将定义和泛化我们的资产函数以采用规范并生成资产 spec specs[0] # take a single spec as reference while buildingasset(namespec[name])
def generic_asset():result requests.get(fwww.donorplatform.org/api/{spec[endpoint]})with open(f{spec[name]}.f{spec[file_type]}, w) as f:if spec[file_type] csv:writer csv.writer(f)writer.writerows(result)elif spec[file_type] json:f.write(result) 最后让我们将通用资产包装在一个函数中该函数将用作生成所有资产的工厂。该函数将采用规范并将其应用于我们的资产。 def generate_donor_platform_asset(spec):asset(namespec[name])def _asset():result requests.get(fwww.donorplatform.org/api/{spec[endpoint]})with open(f{spec[name]}.f{spec[file_type]}, w) as f:if spec[file_type] csv:writer csv.writer(f)writer.writerows(result)elif spec[file_type] json:f.write(result)return _asset 在您的工厂中您只需定义其他规格即可生产更多资产。以下是它在生产中的使用方式 from dagster import Definitions, asset
import requests
import csvspecs [{name: volunteers, endpoint: v1/volunteers, file_type: csv},{name: donations, endpoint: v2/donations, file_type: csv},{name: donors, endpoint: v1/donors, file_type: json},{name: projects, endpoint: v1/projects, file_type: json},{name: fundraisers, endpoint: v1/fundraisers, file_type: csv},
]def generate_donor_platform_asset(spec):asset(namespec[name])def _asset():result requests.get(fwww.donorplatform.org/api/{spec[endpoint]})with open(f{spec[name]}.f{spec[file_type]}, w) as f:if spec[file_type] csv:writer csv.writer(f)writer.writerows(result)elif spec[file_type] json:f.write(result)return _assetdefs Definitions(assets[generate_donor_platform_asset(spec) for spec in specs])如果您运行 您将可以访问 dagster 的 UI 和 localhost3000 上的资产图。当您向列表中添加更多规范并重新加载定义时您将看到生成的更多资产。dagster dev 这是 Dagster 中资产工厂入门的基本数据工程示例。随着您深入研究 Dagster您可以探索更高级的功能。Dagster还提供与许多数据库和数据系统的集成。这使得使用资源工厂以各种格式和位置生成资产变得容易。可以使用资产工厂在云存储系统上生成文件、执行 SQL 或训练机器学习模型。 因此python项目中的资产工厂提供了相同的三个好处 声明资源工厂允许您指定要生产的内容而无需编写如何生成它的详细逻辑可 重用由于资源工厂是模板因此可以在不同的管道或项目中重复使用它们一致使用资源工厂可确保以一致的方式生成资产无论它们在何处或如何使用 七、结论 工厂模式是数据工程师的宝贵工具尤其是在处理不同类型的数据源或复杂对象创建时。它简化了您的代码使其更具可重用性、可扩展性和可维护性。 作为最流行的编程语言之一在使用 Python 进行数据工程时工厂模式非常有用。虽然这些也适用于其他语言但您最有可能遇到 Python 作为通用编程语言落后于大多数流行的数据工程任务工具。
