0

我对管道测试寄予厚望。

我有一个类型的数据框批次:- great_expectations.dataset.pandas_dataset.PandasDataset

我想建立动态验证表达式

batch.("columnname","value") 其中validationtype columname 和value 来自json 文件。

JSON结构:-

{
            "column_name": "sex",
            "validation_type": "expect_column_values_to_be_in_set",
            "validation_value": ["MALE","FEMALE"]
        },

当我构建此表达式时,会收到如下所述的错误消息。

代码:-

def add_validation(self,batch,validation_list):
             for d in validation_list:
                 expression = "." + d["validation_type"] + "(" + d["column_name"] + "," + 
                             str(d["validation_value"]) + ")"
                 print(expression)
                 batch+expression
                 batch.save_expectation_suite(discard_failed_expectations=False)
                 return batch

输出:-

print statement output
.expect_column_values_to_be_in_set(sex,['MALE','FEMALE'])

错误:-

TypeError: ufunc 'add' 不包含签名匹配类型 dtype('


来自标准输入的 Python 3 psycopg2 COPY 失败:.read() 中的错误

我正在尝试应用在页面上找到的代码,特别是目录的“从字符串迭代器复制数据”部分,但我的代码遇到了问题。

由于并非所有来自生成器的行(此处为 log_lines)都可以导入 PostgreSQL 数据库,因此我尝试使用 itertools.filterfalse 过滤正确的行(此处为行),如下面的代码块所示:

def copy_string_iterator(connection, log_lines) -> None:
with connection.cursor() as cursor:
    create_staging_table(cursor)
    log_string_iterator = StringIteratorIO((
        '|'.join(map(clean_csv_value, (                    
                row['date'],
                row['time'],
                row['cs_uri_query'],
                row['s_contentpath'],
                row['sc_status'],
                row['s_computername'],
                ...
                row['sc_substates'],
                row['s_port'],
                row['cs_version'],
                row['c_protocol'],
                row.update({'cs_cookie':'x'}),
                row['timetakenms'],
                row['cs_uri_stem'],
            ))) + '\n')
        for row in filterfalse(lambda line: "#" in line.get('date'), log_lines)
    )
    cursor.copy_from(log_string_iterator, 'log_table', sep = '|')

当我运行它时, cursor.copy_from() 给了我以下错误:

QueryCanceled: COPY from stdin failed: error in .read() call
CONTEXT:  COPY log_table, line 112910

我明白为什么会出现这个错误,这是因为在我使用的测试文件中,只有 112909 行满足 filterfalse 条件。但是为什么它会尝试复制第 112910 行并抛出错误而不是停止呢?

4

1 回答 1

1

在 great_expectations 中,该expectation_suite对象旨在捕获评估期望所需的所有信息。因此,在您的情况下,最自然的做法是将您拥有的源 json 文件转换为 great_expectations 期望套件格式。

做到这一点的最佳方法将取决于您从何处获得原始 JSON 结构——理想情况下,您希望尽早进行翻译(甚至在创建源 JSON 之前?)并保持期望GE 格式。

例如,如果您的所有期望都是 type expect_column_values_to_be_in_set,您可以进行直接翻译:

expectations = []
for d in validation_list:
  expectation_config = {
    "expectation_type": d["validation_type"],
    "kwargs": {
      "column": d["column_name"],
      "value_set": d["validation_value"]
    }
  }
expectation_suite = {
  "expectation_suite_name": "my_suite",
  "expectations": expectations
}

另一方面,如果您正在处理各种不同的期望,您还需要确保 JSON 中的 validation_value 映射到期望的正确 kwargs(例如,如果expect_column_values_to_be_between您确实需要提供min_value和/或max_value)。

于 2020-04-29T13:39:57.513 回答