单一位置的Spark模式管理

编程问答 2022-08-17

问题描述

问题

什么是管理Spark表的架构的最佳方法？您是否看到选项2的任何缺点？您可以提出其他更好的选择吗？

我看到的解决方案

选项1：为代码和元存储保留单独的定义

此方法的缺点是您必须使它们不断保持同步（容易出错）。另一个缺点-如果表具有500列，则会变得很麻烦。

create_some_table.sql [第一个定义] 

-- Databricks Syntax (internal metastore)
CREATE TABLE IF NOT EXISTS some_table (
  Id int,Value string,...
  Year int
)
USING PARQUET
PARTITION BY (Year)
OPTIONS (
  PATH 'abfss://...'
)

some_job.py [第二个定义] 

def run():
   df = spark.read.table('input_table')  # 500 columns
   df = transorm(df)
   # this logic should be in `transform`,but anycase it should be
   df = df.select(
     'Id','Year',F.col('Value').cast(StringType()).alias('Value')  # actually another schema deFinition: you have to enumerate all output columns
   )
   df.write.saveAsTable('some_table')

test_some_job.py [第三个定义] 

def test_some_job(spark):
   output_schema = ...  # another deFinition
   expected = spark.createDataFrame([...],output_schema)

选项2：代码（StructType）中仅保留一个定义

可以即时生成模式。这种方法的好处-可以在一个地方实现简单性和模式定义。 您看到任何弊端吗？ 

def run(input: Table,output: Table):
   df = spark.read.table(input.name)
   df = transform(df)
   save(df,output)    

def save(df: DataFrame,table: Table): 
    df \
        .select(table.schema.fieldNames()) \
        .write \
        .partitionBy(table.partition_by) \
        .option('path',table.path) \
        .saveAsTable(table.name)
    # In case table doesn't exists,Databricks will automatically generate table deFinition
        
class Table(NamedTuple):
    name: str
    path: str
    partition_by: List[str]
    schema: StructType

解决方法

首先让我提出几点，然后提出建议。

  1. 数据的寿命比代码更长。
  2. 上述代码是创建和写入数据的代码，还有一些代码可以读取和使用需要考虑的数据。
  3. 有第三个选项，将数据的定义（模式）与数据一起存储。通常称为“自我描述格式”
  4. 数据的结构会随时间变化。
  5. 该问题被标记为databricks和aws-glue
  6. 镶木地板是逐个文件地自我描述。
  7. Delta Lake表使用镶木地板数据文件，但另外将架构嵌入事务日志中，因此对整个表和架构进行了版本控制。
  8. 数据需要由广泛的工具生态系统使用，因此数据必须是可发现的，架构不应锁定在一个计算引擎中。

推荐：

  1. 以开放格式存储包含数据的架构
  2. 使用Delta Lake格式（结合了镶木地板和交易记录）
  3. 将USING PARQUET更改为USING DELTA
  4. 将您的元存储指向AWS Glue Catalog，Glue目录将存储表名称和位置
  5. 消费者将从Delta Lake表事务日志中解析模式
  6. 架构可以随着编写者代码的发展而发展。

结果：

  1. 您的作者创建模式，并且可以可选地发展模式
  2. 所有使用者都可以在Delta Lake（具体为_delta_log目录）中找到架构（与表版本配对）
,

在我公司，我们有一个非常小的团队，因此我们建立了一个基于C＃模板的生成器，该生成器从SQL Server数据库中读取信息并生成我们的所有笔记本。生成整个系统的DDL模式和处理代码所需的所有信息都在该数据库中。现在，我们唯一手动编写的代码是Dims and Facts加载过程的一部分。所有样板代码都被抽象掉了。对于新表，我们进入元数据库，输入新表和列信息并运行生成器。这将导致您在结果代码中讨论重复项，但是现在我们仅维护元数据数据库信息，而不必查找所有需要更新的位置。如果您已经建立了一个系统，那么这种方法将很难适应，而且说服高层管理人员可能会面临挑战，因为前期还有很多开发工作。但是，如果您从头开始使用咒语自动化，自动化自动化并逐步进行如何使它通用的方法，您将获得解决问题和稳定平台的无与伦比的能力。我在野外看到的唯一能够在大数据空间中执行类似操作的产品是Wherescape，如果我没记错的话，每年每座35,000美元的价格非常昂贵。因此，如果您财大气粗，我强烈建议您看看。 [https://www.wherescape.com/][1] 可能还有其他人，但这是我所知道的唯一一个。

此方法如何带来好处的几个示例。

  1. 我们自己和彼此都有两个开发人员，我们能够在4个月内将130个阶段和40个Dim和Facts从U-SQL Datalake Analytics完全切换到Databricks。逐步关闭所有生成的进程。

  2. 可以说，在加载暂存表时我们发现了一个处理问题，我们更新了流程模板，重新生成了笔记本并重新部署。现在，可能需要花费数周的时间来修复和测试其他日常工作的所有130张表的流程现在都进行了更新。

apache-sparkaws-gluedatabrickshive