无论如何比较两个avro文件,看看数据中存在哪些差异？

Question

理想情况下,我想要像SAS proc比较包装的东西可以给我:

• 每个数据集的行数

• 存在于一个数据集中但不存在于另一个数据集中的行数

• 存在于一个数据集中但不存在于另一个数据集中的变量

• 两个文件中的格式不同的变量(我意识到AVRO文件很少见,但有助于快速了解而无需解密错误)

• 每列不匹配行的总数,以及列的所有不匹配的表示或任何20个不匹配(以最小者为准)

我已经找到了一种方法来确保数据集是等价的,但效率非常低.让我们假设我们有两个avro文件,包含100行和5列(一个键和四个浮点功能).如果我们连接表并创建新变量,这些变量是数据集中匹配要素之间的差异,那么任何非零差异都是数据中的一些不匹配.从那里可以很容易地确定上面的整个要求列表,但似乎可能有更有效的方法.

Answer 1

AVRO文件分别存储架构和数据.这意味着除了带有数据的AVRO文件旁边你应该有一个模式文件,通常它就像*.avsc.这样,您的任务可分为3个部分:

1. 比较架构.这样,您可以获取这些文件中具有不同数据类型的字段,具有不同的字段集等等.这个任务非常简单,可以在Hadoop之外完成,例如在Python中:

   import json
   schema1 = json.load(open('schema1.avsc'))
   schema2 = json.load(open('schema2.avsc'))
   def print_cross (s1set, s2set, message):
       for s in s1set:
           if not s in s2set:
               print message % s
   s1names = set( [ field['name'] for field in schema1['fields'] ] )
   s2names = set( [ field['name'] for field in schema2['fields'] ] )
   print_cross(s1names, s2names, 'Field "%s" exists in table1 and does not exist in table2')
   print_cross(s2names, s1names, 'Field "%s" exists in table2 and does not exist in table1')
   def print_cross2 (s1dict, s2dict, message):
       for s in s1dict:
           if s in s2dict:
               if s1dict[s] != s2dict[s]:
                   print message % (s, s1dict[s], s2dict[s])
   s1types = dict( zip( [ field['name'] for field in schema1['fields'] ],  [ str(field['type']) for field in schema1['fields'] ] ) )
   s2types = dict( zip( [ field['name'] for field in schema2['fields'] ],  [ str(field['type']) for field in schema2['fields'] ] ) )
   print_cross2 (s1types, s2types, 'Field "%s" has type "%s" in table1 and type "%s" in table2')
   

这是一个模式的例子:

{"namespace": "example.avro",
 "type": "record",
 "name": "User",
 "fields": [
     {"name": "name", "type": "string"},
     {"name": "favorite_number",  "type": ["int", "null"]},
     {"name": "favorite_color", "type": ["string", "null"]}
 ]
}

{"namespace": "example.avro",
 "type": "record",
 "name": "User",
 "fields": [
     {"name": "name", "type": "string"},
     {"name": "favorite_number",  "type": ["int"]},
     {"name": "favorite_color", "type": ["string", "null"]},
     {"name": "test", "type": "int"}
 ]
}

这是输出:

[localhost:temp]$ python compare.py 
Field "test" exists in table2 and does not exist in table1
Field "favorite_number" has type "[u'int', u'null']" in table1 and type "[u'int']" intable2

2. 如果模式相同(如果模式不相等,您可能不需要比较数据),那么您可以通过以下方式进行比较.匹配任何情况的简单方法:为每个行计算md5哈希,根据此md5哈希的值连接两个表.如果将为您提供两个表中相同的行数,table1特定的行数和table2特定的行数.它可以在Hive中轻松完成,这是MD5 UDF的代码:https://gist.github.com/dataminelab/1050002

3. 要比较字段到字段,您必须知道表的主键,并在主键上连接两个表,比较字段

以前我已经开发了表的比较函数,它们通常看起来像这样:

1. 检查两个表是否存在且可用
2. 比较他们的架构.如果架构中存在一些错配 - 中断
3. 如果指定了主键:
   1. 使用完全外连接加入主键上的两个表
   2. 计算每行的md5哈希值
   3. 输出具有诊断的主键(PK仅存在于table1中,PK仅存在于table2中,PK存在于两个表中但数据不匹配)
   4. 获取每个有问题的类的100行相同,连接两个表并输出到"mistmatch example"表中
4. 如果未指定主键:
   1. 计算每行的md5哈希值
   2. table1与md5hash值上的table2的完全外连接
   3. 计数匹配行数,仅在table1中存在行数,仅在table2中存在行数
   4. 获取每个错配类型的100行样本并输出到"mistmatch example"表

通常开发和调试这样的功能需要4-5个工作日