python - Azure 和 Databricks 寄予厚望

Question

我想针对 ADLS Gen2 中的 csv 文件运行 Great_expectation 测试套件。在我的 ADLS 上，我有一个名为“input”的容器，其中有一个文件位于 input/GE/ind.csv。我使用 InferredAssetAzureDataConnector。我能够创建并测试/验证数据源配置。但是当我验证我的数据时，我遇到了以下错误。

import datetime
import pandas as pd
from ruamel import yaml

from great_expectations.core.batch import RuntimeBatchRequest
from great_expectations.data_context import BaseDataContext
from great_expectations.data_context.types.base import (
    DataContextConfig,
    FilesystemStoreBackendDefaults,
)

from ruamel import yaml
import great_expectations as ge
from great_expectations.core.batch import Batch, BatchRequest


#Root Directory
root_directory = "/dbfs/FileStore/great_expectation_official/"

#Data Context
data_context_config = DataContextConfig(
    store_backend_defaults=FilesystemStoreBackendDefaults(
        root_directory=root_directory
    ),
)
context = BaseDataContext(project_config=data_context_config)


#Configure your Datasource
datasource_config = {
    "name": "my_azure_datasource",
    "class_name": "Datasource",
    "execution_engine": {
        "class_name": "SparkDFExecutionEngine",
        "azure_options": {
            "account_url": "https://<account_Name>.blob.core.windows.net",
            "credential": "ADLS_key",
        },
    },
    "data_connectors": {
        "default_inferred_data_connector_name": {
            "class_name": "InferredAssetAzureDataConnector",
            "azure_options": {
                "account_url": "https://<account_Name>.blob.core.windows.net",
                "credential": "ADLS_key",
            },
            "container": "input",
            "name_starts_with": "/GE/",
            "default_regex": {
                "pattern": "(.*)\\.csv",
                "group_names": ["data_asset_name"],
            },
        },
    },
}

context.test_yaml_config(yaml.dump(datasource_config))

context.add_datasource(**datasource_config)

batch_request = BatchRequest(
    datasource_name="my_azure_datasource",
    data_connector_name="default_inferred_data_connector_name",
    data_asset_name="data_asset_name",
    batch_spec_passthrough={"reader_method": "csv", "reader_options": {"header": True}},
)

context.create_expectation_suite(
    expectation_suite_name="test_suite", overwrite_existing=True
)
validator = context.get_validator(
    batch_request=batch_request, expectation_suite_name="test_suite"
)

[Error_snapshot_click_here]

[csv_data_snapshot]

有人可以帮我找出问题所在吗？

Answer 1

您可以通过以下代码检查您的批处理列表是否确实为空。

context.get_batch_list(batch_request=batch_request)

如果这是空的，则您的 data_asset_names 可能有问题。您可以检查以下代码的输出中是否使用了正确的数据 Assets 名称。

context.test_yaml_config(yaml.dump(my_spark_datasource_config))

在输出中，有一个可用的 data_asset_names 列表可供您选择。如果您的 BatchRequest 的数据 asset_name 不在列表中，则您将得到一个空的 batch_list，因为该数据 Assets 不可用。这里应该有一个警告，但我认为它没有实现。

我也遇到了同样的问题，不过昨天解决了。另外，您应该生成一个可行的错误示例，以便人们可以探索代码。这样就更容易为您提供帮助。

希望这有帮助!