如何在 python 中比较 Databricks 笔记本中的两个模式

Posted 2023-04-15

【中文标题】如何在 python 中比较 Databricks 笔记本中的两个模式

【英文标题】：How to compare two schema in Databricks notebook in python

【发布时间】：2021-12-31 16:45:00

【问题描述】：

我将使用 databricks 笔记本提取数据。我想根据我期望的这些数据的架构来验证摄取的数据架构。

所以基本上我有：

    validation_schema = StructType([
      StructField("a", StringType(), True),
      StructField("b", IntegerType(), False),
      StructField("c", StringType(), False),
      StructField("d", StringType(), False)
    ])

    data_ingested_good = [("foo",1,"blabla","36636"),
     ("foo",2,"booboo","40288"),
     ("bar",3,"fafa","42114"),
     ("bar",4,"jojo","39192"),
     ("baz",5,"jiji","32432")
    ]

    data_ingested_bad = [("foo","1","blabla","36636"),
     ("foo","2","booboo","40288"),
     ("bar","3","fafa","42114"),
     ("bar","4","jojo","39192"),
     ("baz","5","jiji","32432")
    ]
     
    data_ingested_good.printSchema()
    data_ingested_bad.printSchema()
    validation_schema.printSchema()

我见过类似的问题，但答案总是在 scala 中。

【参考方案1】：

这实际上取决于您要比较的模式的确切要求和复杂性 - 例如，忽略可空性标志与考虑它、列的顺序、对映射/结构/数组的支持等。另外，做如果模式匹配或不匹配，您想查看差异或只是一个标志。

在最简单的情况下，它可以像下面这样简单 - 只需比较模式的字符串表示：

def compare_schemas(df1, df2):
  return df1.schema.simpleString() == df2.schema.simpleString()

我个人建议使用现有的库，例如 Chispa，它具有更高级的模式比较功能 - 你可以调整检查，它会显示差异等。安装后（你可以这样做 %pip install chispa） - 这个如果架构不同，将抛出异常：

from chispa.schema_comparer import assert_schema_equality

assert_schema_equality(df1.schema, df2.schema)

【讨论】：

谢谢。这确实是一个复杂的问题。我不知道如何制定它。但是我认为simpleString() 已经是我在第一个基础级别提出的问题的一个很好的答案。

【参考方案2】：

另一种方法，你可以根据简单的pythonlist比较找到区别。

dept = [("Finance",10), 
        ("Marketing",20), 
        ("Sales",30), 
        ("IT",40) 
      ]
deptColumns = ["dept_name","dept_id"]

dept1 = [("Finance",10,'999'), 
        ("Marketing",20,'999'), 
        ("Sales",30,'999'), 
        ("IT",40,'999') 
      ]
deptColumns1 = ["dept_name","dept_id","extracol"]

deptDF = spark.createDataFrame(data=dept, schema = deptColumns)
dept1DF = spark.createDataFrame(data=dept1, schema = deptColumns1)
deptDF_columns=deptDF.schema.names
dept1DF_columns=dept1DF.schema.names

list_difference = []
for item in dept1DF_columns:
  if item not in deptDF_columns:
     list_difference.append(item)

print(list_difference)

屏幕打印：

【讨论】：

如果 deptDF 的列数比 dept1DF 多，它不会发现差异