加速Teradata ETL性能：使用AWS Glue的高级分区技术

加速Teradata ETL性能：使用AWS Glue的先进分区技术 目录 01高级分区技术在AWS Glue中的应用 •加速使用 AWS Glue 提升Teradata ETL 性能•为什么在ETL转换中使用分区？ 02 传统JDBC在大规模数据写入中的挑战 •现实世界中对JDBC局限性的影响•为什么需要更好的解决方案？ 03如何使用AWS Glue应对Hexaware的ETL挑战分区策略 •1. 由Spark驱动的默认分区-代码示例：默认分区-功能：使用 SQLAlchemy 将分区数据插入到 Teradata 表中-函数将PySpark DataFrame基于分区处理插入至Teradata-如何默认批分区工作-默认分区的好处-默认分区限制 •2. 自定义批量分区-功能：批量将分区数据插入Teradata-功能：以自定义分区方式批量将 Spark DataFrame 插入 Teradata 数据库-自定义批量分区是如何工作的-优势在于自定义批量分区-何时使用自定义批分区 • 关于分区策略的关键要点 04比较性能指标 •关键观察-性能影响•为什么这很重要 05最佳实践：优化AWS Glue中分区数据写入的方案 06赛华思与亚马逊云科技战略：利用亚马逊云科技Glue克服传统的ETL挑战 加速Teradata ETL性能：高级分区技术在 AWS Glue 中的应用 高级分区技术在AWSGlue中的应用 今天，企业每天都会产生和处理大量的数据。有效地管理这些数据涌入对于提供及时的业务洞察和最小化资源消耗至关重要。大规模的数据写入可能成为依赖如Teradata等数据库的企业的一个瓶颈，造成延迟、系统过载并增加运营成本。AWS Glue是一种无服务器数据集成服务，通过自动化提取、转换和加载（ETL）工作流程，简化并加速了该过程，实现大规模数据无缝移动和转换。 01 使用AWS Glue加速Teradata ETL性能 在传统ETL方法面对数百万或数十亿行数据、顺序数据处理、资源竞争和错误处理复杂性时常常力不从心，这些是常见的难题。分区，作为一种基本的数据工程技术，通过将更大的数据集划分为逻辑块以实现并行处理来解决这些挑战。 我们的指南探讨了如何通过AWS Glue的分区功能和其PySpark能力，赋能数据工程师在ETL流程优化中，实现向Teradata的高性能数据写入。通过分区大型数据集并利用Glue的分布式架构，您可以实现更快的执行速度、更好的容错性和无缝的可扩展性。 在本指南中，您将学习： •为什么Java数据库连接（JDBC）在处理大规模数据写入时遇到困难。•如何通过AWS Glue进行分区以转换和提升ETL性能•实施分区数据写入Teradata的高效步骤。•关键指标对比 JDBC 和 Glue，突出性能提升。•最佳实践：在AWS Glue中优化分区数据写入 分区是高效数据工程的基石。它将大型数据集拆分为更小、更逻辑的片段，从而实现并行处理。这种简单而强大的技术带来了显著的优势： •改进的性能：更小的数据块并发处理，显著减少了执行时间。•优化资源利用：平衡的工作负载有助于防止内存或CPU瓶颈。•巨大的可扩展性：调整分区策略可以无缝地处理数百万甚至数十亿行。•并行写入：并发写入将负载均匀分布在数据库中，从而提高效率。•错误隔离：处理更小的数据块简化了错误恢复，使故障排除更加容易管理。 对于Teradata来说，分区重新定义了数据如何被摄取： •而不是顺序写入，分区并行写入，利用 AWS Glue 的分布式架构和 Teradata 的批量加载数据的能力。•这种方法提高了执行速度、容错能力和可扩展性。 桥接现代数据需求与传统ETL局限性之间的差距，转换工作流程为更快速、更高效的管道。 将数百万行数据写入像Teradata这样的数据库时，使用传统的JDBC（Java数据库连接）方法会面临重大挑战。虽然JDBC对于小规模操作来说是足够的，但对于大量数据集来说，它成为一个瓶颈。 关键限制及其影响： 传统JDBC在大规模数据写入中的关键局限性 1. 缓慢且顺序处理。JDBC按照行或小批量处理数据，在处理大型数据集时会导致显著延迟。这种顺序方法难以满足现代数据需求，通常会导致关键操作的延迟。 2. 无并行处理 JDBC不允许在多个线程或系统间并行写入数据。它依赖于与数据库的单一连接，因此无法利用现代分布式计算工具如Spark。这种缺乏并行性成为大数据集的主要瓶颈。 3. 高资源使用 在内存中写入大量批数据消耗•ETL系统压力：显著占用CPU和RAM，降低其他任务的性能。 单线程连接涌入数据库。•数据库过载：增加竞争风险和减速的可能性。 4. 错误处理复杂性 处理批量处理中的错误相当繁琐： •一批中若出现单个缺陷通常需要重新加工整个批次，造成时间和精力的浪费。•故障排除此类问题会引入延迟和运营开销。 5. 有限的可扩展性 随着数据量增长，基于JDBC的管道扩展变得越来越困难。尽管存在诸如将数据分成更小块或者增加连接等解决方案，但它们都需要手动干预，并且增加了复杂性，而没有解决核心的低效问题。 6. 数据倾斜 数据倾斜发生在数据在分区中不均匀分布时，导致处理时间不平衡和资源利用效率低下。某些分区可能包含比其他分区显著更多的数据，导致某些Spark任务执行时间更长，而其他任务则快速完成。这种不平衡降低了并行效率，增加了ETL执行时间，并可能导致某些工作节点上的资源瓶颈。 现实世界中对JDBC局限性的影响 局限性转化为： • 延迟：大规模数据写入可能需要数小时，从而延迟了洞察力和决策制定。•高成本：资源使用效率低下导致运营成本增加。• 增加手动工作：重试机制和错误恢复需要持续的监控。 为什么需要更好的解决方案？ 克服这些挑战，解决方案必须： • 通过并行化加快数据处理速度。• 优化资源利用以减轻系统压力。• 优雅处理错误，隔离故障而不影响整个数据集。 AWS Glue，凭借其内置的分区和分布式处理能力，有效地解决了这些问题，实现了可扩展、高效和可靠的ETL管道。 Hexaware利用AWS Glue的分区功能，克服了传统JDBC方法在将大规模数据集写入Teradata时的限制。通过利用Spark的分布式架构，我们设计和部署了可扩展的ETL管道，实现了更快的执行速度、高效的资源利用和容错性。 如何利用AWS Glue解决Hexaware在分区策略中的ETL挑战 03 分区策略在Hexaware的采用 Hexaware根据具体项目需求采用了两种主要的分区策略： 分区策略 1. Spark的默认分区：Spark通过其内部负载均衡机制，将数据分配到其默认分区中。 2. 自定义批量分区：您定义分区数量并控制以块写入数据时的批量大小，从而提高性能和容错能力。 1. 由Spark驱动的默认分区 在此方法中，Spark根据其内部逻辑自动对数据集进行分区，考虑数据集大小、集群资源和配置。每个分区独立处理并写入目标数据库。这种自动化方法平衡了分区，并允许在最小配置下进行并行处理。 代码示例：默认分区 这里是一个Spark默认分区如何高效地将数据写入Teradata的示例： 功能：使用 SQLAlchemy 将分区数据插入到 Teradata 表中 def insert_partition_to_teradata(partition, teradata_table, td_credentials): “”” 将单个数据分区插入到Teradata表中。 参数： 分区（iterable）：待插入行的一个分区。teradata_table（字符串）：目标Teradata表的名字。td_credentials（字典）：Teradata连接凭据。 回报： 整型：从分区插入的行数。 尝试： # 记录插入过程的开始打印(f”正在将分区数据插入到 {teradata_table}...”) # 建立与Teradata的连接 td_engine = create_teradata_engine(td_credentials) Session = sessionmaker(bind=td_engine) session = Session() 将分区转换为字典列表rows = [行.asDict() for 行 in partition] 如果行：从分区首行提取列名columns = rows[0].keys() # 为参数化SQL查询创建一个占位符字符串占位符 = “, “.join([f”:{列}” for 列 in 列表]) )# 准备SQL INSERT语句query = text(f”插入到 {teradata_table} ({‘, ‘.join(columns)}) “f”值({占位符})” 执行分区中的行查询会话执行查询（query），并将结果行（rows）返回。会话提交()打印(f\"从该分区插入 {len(rows)} 行。\") 返回插入的行数 # Return the number of rows insertedelse:处理分区中无数据的情况print(“此分区中无数据可插入。”)返回 0 除了异常作为 e：处理在插入过程中发生的任何错误打印（f“插入分区时出错：{e}”）返回 0 最后：finally:确保会话已关闭，以避免资源泄漏会话关闭() 函数将PySpark DataFrame基于分区处理插入至Teradata def insert_data_into_teradata(df, teradata_table, td_credentials): “”” 在Teradata中插入数据的函数，参数包括数据框(df)、表名(teradata_table)和Teradata凭据(td_credentials)。将PySpark DataFrame插入到Teradata中，使用Spark的默认分区。 参数： df (DataFrame)：要插入到Teradata的PySpark DataFrame。teradata_table (str): 目标Teradata表的名称。td_credentials (dict): Teradata 连接凭据。 int: 插入到Teradata表中的总行数。 定义一个用于处理单个分区的函数 def process_partition(partition): 通过将每个分区插入到Teradata中对其进行处理。 参数：分区（可迭代对象）：一行数据的单个分区。 “””回报：列表：包含从分区中插入的行数的列表。 # 调用分区插入函数并返回计数返回：[插入分区到Teradata（分区teradata_tabletd_credentials )] )打印(“开始向 {teradata_table} 插入数据”使用默认分区... 计算插入的总行数总插入计数 = insert_counts之和 )# 记录最终结果打印(f”总计插入到 {teradata_table} 中的行数：“f\"{total_insert_count}\ # 返回插入行的总数返回总插入计数 默认批量分区工作原理 1. 分区处理：a. Spark自动根据集群配置将数据集分割成分区。b.每个分区是数据的一个子集，Spark并行处理这些子集。 2. 处理每个分区：a. insert_partition_to_teradata() 函数负责将每个分区写入 Teradata。 b.它将分区转换为字典列表（行），并准备SQL语句INSERT查询。分区的行在一个单独的事务中写入，确保原子性。 3. 并行执行： a. Spark的mapPartitions()函数将insert_partition_to_teradata()函数应用于所有分区，实现并发处理。b. 数据库负