Dagster集成Great Expectations遇'SolidDefinition'不可下标错误求助

使用Dagster 0.15.9集成Great Expectations 0.15.44时，调用ge_validation_op_factory后尝试通过expectation["success"]访问验证结果，触发TypeError: 'SolidDefinition' object is not subscriptable错误。

核心原因是**ge_validation_op_factory的作用是生成Op的定义对象（SolidDefinition），而非直接执行验证并返回ExpectationResult**。你错误地在自定义Op内部调用该工厂函数，返回的是Op定义而非验证执行结果，自然无法用下标访问属性。

将ge_validation_op_factory的调用移到Op定义外部，生成独立的验证Op；然后在graph中编排这个Op的执行，获取其返回的验证结果。

import json
import boto3
import dagster_ge
from dagster import (
    op,
    graph,
    Field,
    String,
    OpExecutionContext,
)

from typing import List, Dict


@op(
    config_schema={
        "bucket": Field(
            String,
            description="s3 bucket name",
        ),
        "path_in_s3": Field(
            String,
            description="Prefix representing the path to data",
        ),
        "technical_date": Field(
            String,
            description="date string to fetch data",
        ),
        "file_name": Field(
            String,
            description="file name that contains the data",
        ),
    }
)
def read_in_json_datafile_from_s3(context: OpExecutionContext):
    bucket = context.op_config["bucket"]
    path_in_s3 = context.op_config["path_in_s3"]
    technical_date = context.op_config["technical_date"]
    file_name = context.op_config["file_name"]

    object = f"{path_in_s3}/technical_date={technical_date}/{file_name}"

    s3 = boto3.resource("s3")

    content_object = s3.Object(bucket, object)
    file_content = content_object.get()["Body"].read().decode("utf-8")
    json_content = json.loads(file_content)

    return json_content


@op
def process_example_dq(data: List[Dict]):
    return len(data)


@op
def postprocess_example_dq(numrows, expectation):
    if expectation["success"]:
        return numrows
    else:
        raise ValueError("Great Expectations验证失败")

# 把ge_validation_op_factory移到外部，生成独立的验证Op
ge_validation_op = dagster_ge.ge_validation_op_factory(
    name='ge_validation_op',
    datasource_name='my_bucket.data_pipelines.raw_data.example_update',
    suite_name='suite.data_pipelines.raw_data.example_update',
)


@graph(
    config={
        "read_in_json_datafile_from_s3": {
            "config": {
                "bucket": "my_bucket",
                "path_in_s3": "my_path",
                "technical_date": "2023-01-24",
                "file_name": "myfile_20230124.json",
            }
        },
    },
)
def example_update_evaluation():
    output_dict = read_in_json_datafile_from_s3()
    nb_items = process_example_dq(data=output_dict)
    # 直接调用生成的验证Op，获取执行结果
    expectation_result = ge_validation_op()
    postprocess_example_dq(
        numrows=nb_items,
        expectation=expectation_result,
    )

• 如果GE验证需要依赖前面Op输出的数据，可使用dagster_ge.ge_validation_op_factory的batch_identifiers参数关联数据，或通过ge_validation_op的输入传递数据。
• 需确保Great Expectations的数据源、suite配置正确，否则验证Op执行时会出现数据源或suite不存在的错误。

内容的提问来源于stack exchange，提问作者Imad