Version: 0.15.50

How to connect to data on S3 using Spark

This guide will help you connect to your data stored on AWS S3 using Spark. This will allow you to ValidateThe act of applying an Expectation Suite to a Batch. and explore your data.

Prerequisites

This guide assumes you have:

Completed the Getting Started Tutorial
Access to data on an AWS S3 bucket
Access to a working Spark installation

Steps

1. Choose how to run the code in this guide

Get an environment to run the code in this guide. Please choose an option below.

CLI + filesystem
No CLI + filesystem
No CLI + no filesystem

If you use the Great Expectations CLICommand Line Interface, run this command to automatically generate a pre-configured Jupyter Notebook. Then you can follow along in the YAML-based workflow below:

great_expectations datasource new

2. Instantiate your project's DataContext

Import these necessary packages and modules.

from ruamel import yaml

import great_expectations as gx
from great_expectations.core.batch import Batch, BatchRequest, RuntimeBatchRequest

Load your DataContext into memory

Use one of the guides below based on your deployment:

Please proceed only after you have instantiated your DataContext.

3. Configure your Datasource

Using this example configuration, add in your S3 bucket and path to a directory that contains some of your data:

YAML
Python

name: my_s3_datasource
class_name: Datasource
execution_engine:
    class_name: SparkDFExecutionEngine
data_connectors:
    default_runtime_data_connector_name:
        class_name: RuntimeDataConnector
        batch_identifiers:
            - default_identifier_name
    default_inferred_data_connector_name:
        class_name: InferredAssetS3DataConnector
        bucket: <your_s3_bucket_here>
        prefix: <bucket_path_to_data>
        default_regex:
            pattern: (.*)\.csv
            group_names:
                - data_asset_name

Run this code to test your configuration.

context.test_yaml_config(datasource_yaml)

datasource_config = {
    "name": "my_s3_datasource",
    "class_name": "Datasource",
    "execution_engine": {"class_name": "SparkDFExecutionEngine"},
    "data_connectors": {
        "default_runtime_data_connector_name": {
            "class_name": "RuntimeDataConnector",
            "batch_identifiers": ["default_identifier_name"],
        },
        "default_inferred_data_connector_name": {
            "class_name": "InferredAssetS3DataConnector",
            "bucket": "<your_s3_bucket_here>",
            "prefix": "<bucket_path_to_data>",
            "default_regex": {
                "pattern": "(.*)\\.csv",
                "group_names": ["data_asset_name"],
            },
        },
    },
}

Run this code to test your configuration.

context.test_yaml_config(yaml.dump(datasource_config))

If you specified an S3 path containing CSV files you will see them listed as Available data_asset_names in the output of test_yaml_config().

Feel free to adjust your configuration and re-run test_yaml_config() as needed.

4. Save the Datasource configuration to your DataContext

Save the configuration into your DataContext by using the add_datasource() function.

YAML
Python

context.add_datasource(**yaml.load(datasource_yaml))

context.add_datasource(**datasource_config)

5. Test your new Datasource

Verify your new DatasourceProvides a standard API for accessing and interacting with data from a wide variety of source systems. by loading data from it into a ValidatorUsed to run an Expectation Suite against data. using a Batch RequestProvided to a Datasource in order to create a Batch..

Specify an S3 path to single CSV
Specify a data_asset_name

Add the S3 path to your CSV in the path key under runtime_parameters in your RuntimeBatchRequest.

tip

The path you will want to use is your S3 URI, not the URL.

batch_request = RuntimeBatchRequest(
    datasource_name="version-0.15.50 my_s3_datasource",
    data_connector_name="version-0.15.50 default_runtime_data_connector_name",
    data_asset_name="version-0.15.50 <your_meangingful_name>",  # this can be anything that identifies this data_asset for you
    runtime_parameters={"path": "<path_to_your_data_here>"},  # Add your S3 path here.
    batch_identifiers={"default_identifier_name": "default_identifier"},
)

Then load data into the Validator.

context.add_or_update_expectation_suite(expectation_suite_name="version-0.15.50 test_suite")
validator = context.get_validator(
    batch_request=batch_request, expectation_suite_name="version-0.15.50 test_suite"
)
print(validator.head())

Add the name of the Data AssetA collection of records within a Datasource which is usually named based on the underlying data system and sliced to correspond to a desired specification. to the data_asset_name in your BatchRequest.

batch_request = BatchRequest(
    datasource_name="version-0.15.50 my_s3_datasource",
    data_connector_name="version-0.15.50 default_inferred_data_connector_name",
    data_asset_name="version-0.15.50 <your_data_asset_name>",
    batch_spec_passthrough={"reader_method": "csv", "reader_options": {"header": True}},
)

Then load data into the Validator.

context.add_or_update_expectation_suite(expectation_suite_name="version-0.15.50 test_suite")
validator = context.get_validator(
    batch_request=batch_request, expectation_suite_name="version-0.15.50 test_suite"
)
print(validator.head())

🚀🚀 Congratulations! 🚀🚀 You successfully connected Great Expectations with your data.

Additional Notes

To view the full scripts used in this page, see them on GitHub:

Next Steps

Now that you've connected to your data, you'll want to work on these core skills:

This guide assumes you have:

Steps​

1. Choose how to run the code in this guide​

2. Instantiate your project's DataContext​

3. Configure your Datasource​

4. Save the Datasource configuration to your DataContext​

5. Test your new Datasource​

Additional Notes​

Next Steps​