feluelle commented on a change in pull request #11113:
URL: https://github.com/apache/airflow/pull/11113#discussion_r494205186
##########
File path: airflow/providers/google/cloud/operators/bigquery.py
##########
@@ -2109,3 +2128,265 @@ def execute(self, context: Any):
def on_kill(self):
if self.job_id and self.cancel_on_kill:
self.hook.cancel_job(job_id=self.job_id,
project_id=self.project_id, location=self.location)
+
+
+class GreatExpectationsBigQueryOperator(BaseOperator):
+ """
+ Use Great Expectations to validate data expectations against a
BigQuery table or the result of a SQL query.
+ The expectations need to be stored in a JSON file sitting in an
accessible GCS bucket. The validation results
+ are output to GCS in both JSON and HTML formats.
+ Here's the current list of expectations types:
+
https://docs.greatexpectations.io/en/latest/reference/glossary_of_expectations.html
+ Here's how to create expectations files:
+
https://docs.greatexpectations.io/en/latest/guides/tutorials/how_to_create_expectations.html
+ :param gcp_project: The GCP project which houses the GCS buckets
where the expectations files are stored
+ and where the validation files & data docs will be output (e.g.
HTML docs showing if the data matches
+ expectations).
+ :type gcp_project: str
+ :param expectations_file_name: The name of the JSON file containing
the expectations for the data.
+ :type expectations_file_name: str
+ :param gcs_bucket: Google Cloud Storage bucket where expectation
files are stored and where validation outputs
+ and data docs will be saved.
+ (e.g.
gs://<gcs_bucket>/<gcs_expectations_prefix>/<expectations_file_name>
+ gs://mybucket/myprefix/myexpectationsfile.json )
+ :type gcs_bucket: str
+ :param gcs_expectations_prefix: Google Cloud Storage prefix where the
expectations file can be found.
+ (e.g. 'ge/expectations')
+ :type gcs_expectations_prefix: str
+ :param gcs_validations_prefix: Google Cloud Storage prefix where the
validation output files should be saved.
+ (e.g. 'ge/validations')
+ :type gcs_validations_prefix: str
+ :param gcs_datadocs_prefix: Google Cloud Storage prefix where the
validation datadocs files should be saved.
+ (e.g. 'ge/datadocs')
+ :type gcs_datadocs_prefix: str
+ :param validation_type: For the set of data to be validated (i.e.
compared against expectations), is it already
+ sitting in a BigQuery table or do you want to validate the data
returned by a SQL query? The options are
+ 'TABLE' or 'SQL'.
+ :type validation_type: str
+ :param validation_type_input: The name of the BigQuery table
(dataset_name.table_name) if the validation_type
+ is 'TABLE' or the SQL query string if the validation_type is 'SQL'.
+ :type validation_type_input: str
+ :param bigquery_conn_id: Name of the BigQuery connection that contains
the connection and credentials
+ info needed to connect to BigQuery.
+ :type bigquery_conn_id: str
+ :param bq_dataset_name: The name of the BigQuery data set where any
temp tables will be created that are needed
+ as part of the GE validation process.
+ :type bq_dataset_name: str
+ :param send_alert_email: Send an alert email if one or more
expectations fail to be met? Defaults to True.
+ :type send_alert_email: boolean
+ :param include_datadocs_link_in_email: Include in the alert email a
link to the data doc in GCS that shows the
+ validation results? Defaults to False because there's extra setup
needed to serve HTML data docs stored in
+ GCS. When set to False, only a GCS path to the results are
included in the email.
+ Execute the following steps if you want a clickable link for the
data doc to be included in the
+ email:
+ 1) Set up a GAE app to serve the data docs by following the
instructions here:
+
https://docs.greatexpectations.io/en/latest/guides/how_to_guides/configuring_data_docs/how_to_host_and_share_data_docs_on_gcs.html
+ 2) Create an Airflow variable called
'great_expectations_datadocs_domain' with a value set to the domain
+ google creates in step 1 to serve the data docs (e.g.
'ge-data-docs-dot-my-gcp-project.ue.r.appspot.com').
+ This operator will look for it if
include_datadocs_link_in_email=True
+ :type include_datadocs_link_in_email: boolean
+ :param email_to: Email address to receive any alerts when
expectations are not met.
+ :type email_to: str
+ :param fail_if_expectations_not_met: Fail the Airflow task if
expectations are not met? Defaults to True.
+ :type fail_if_expectations_not_met: boolean
+ """
+
+ @apply_defaults
+ def __init__(self, gcp_project, expectations_file_name, gcs_bucket,
gcs_expectations_prefix,
+ gcs_validations_prefix, gcs_datadocs_prefix, validation_type,
validation_type_input,
+ bq_dataset_name, email_to, send_alert_email=True,
include_datadocs_link_in_email=False,
+ fail_if_expectations_not_met=True,
+ bigquery_conn_id='bigquery_default',
+ *args, **kwargs):
+ self.expectations_file_name = expectations_file_name
+ if validation_type.upper() not in GreatExpectationsValidations:
+ raise AirflowException("argument 'validation_type' must be one of
%r." % great_expectations_valid_type)
+ self.validation_type = validation_type
+ self.validation_type_input = validation_type_input
+ self.bigquery_conn_id = bigquery_conn_id
+ self.bq_dataset_name = bq_dataset_name
+ self.gcp_project = gcp_project
+ self.gcs_bucket = gcs_bucket
+ self.gcs_expectations_prefix = gcs_expectations_prefix
+ self.gcs_validations_prefix = gcs_validations_prefix
+ self.gcs_datadocs_prefix = gcs_datadocs_prefix
+ self.email_to = email_to
+ self.send_alert_email = send_alert_email
+ self.include_datadocs_link_in_email = include_datadocs_link_in_email
+ self.fail_if_expectations_not_met = fail_if_expectations_not_met
+
+ super(GreatExpectationsBigQueryOperator, self).__init__(*args,
**kwargs)
+
+ def execute(self, context):
+ conn = BaseHook.get_connection(self.bigquery_conn_id)
+ connection_json = json.loads(conn.extra)
+
+ project_config = DataContextConfig(
+ config_version=2,
+ datasources={
+ "bq_datasource": {
+ "credentials": {
+ "url": "bigquery://" + connection_json[
+ 'extra__google_cloud_platform__project'] + "/" +
self.bq_dataset_name + "?credentials_path=" +
+
connection_json['extra__google_cloud_platform__key_path']
+ },
+ "class_name": "SqlAlchemyDatasource",
+ "module_name": "great_expectations.datasource",
+ "data_asset_type": {
+ "module_name": "great_expectations.dataset",
+ "class_name": "SqlAlchemyDataset"
+ }
+ }
+ },
+ expectations_store_name="expectations_GCS_store",
+ validations_store_name="validations_GCS_store",
+ evaluation_parameter_store_name="evaluation_parameter_store",
+ plugins_directory=None,
+ validation_operators={
+ "action_list_operator": {
+ "class_name": "ActionListValidationOperator",
+ "action_list": [
+ {
+ "name": "store_validation_result",
+ "action": {"class_name":
"StoreValidationResultAction"},
+ },
+ {
+ "name": "store_evaluation_params",
+ "action": {"class_name":
"StoreEvaluationParametersAction"},
+ },
+ {
+ "name": "update_data_docs",
+ "action": {"class_name": "UpdateDataDocsAction"},
+ },
+ ],
+ }
+ },
+ stores={
+ 'expectations_GCS_store': {
+ 'class_name': 'ExpectationsStore',
+ 'store_backend': {
+ 'class_name': 'TupleGCSStoreBackend',
+ 'project': self.gcp_project,
+ 'bucket': self.gcs_bucket,
+ 'prefix': self.gcs_expectations_prefix
+ }
+ },
+ 'validations_GCS_store': {
+ 'class_name': 'ValidationsStore',
+ 'store_backend': {
+ 'class_name': 'TupleGCSStoreBackend',
+ 'project': self.gcp_project,
+ 'bucket': self.gcs_bucket,
+ 'prefix': self.gcs_validations_prefix
+ }
+ },
+ "evaluation_parameter_store": {"class_name":
"EvaluationParameterStore"},
+ },
+ data_docs_sites={
+ "GCS_site": {
+ "class_name": "SiteBuilder",
+ "store_backend": {
+ "class_name": "TupleGCSStoreBackend",
+ "project": self.gcp_project,
+ "bucket": self.gcs_bucket,
+ 'prefix': self.gcs_datadocs_prefix
+ },
+ "site_index_builder": {
+ "class_name": "DefaultSiteIndexBuilder",
+ },
+ }
+ },
+ config_variables_file_path=None,
+ commented_map=None,
+ )
Review comment:
Yes Tomek, I already played around with GE for some time but never
directly used the `DataContextConfig`.
This might be a good option to use instead of having to build a whole file
structure through `great_expectations init` and having to manage those yaml
files. I can see that work, but I would give the caller of this operator more
control about the sections like datasources, validation_operators, stores and
data_docs_sites.
I've already read @dlamblin comment on adding more "abstraction layers" when
we want to introduce GE to the project. I am all for it. I think this would
make the integration in for example BigQuery a lot easier if we have such a
base GE operator.
##########
File path: airflow/providers/google/cloud/operators/bigquery.py
##########
@@ -2109,3 +2128,265 @@ def execute(self, context: Any):
def on_kill(self):
if self.job_id and self.cancel_on_kill:
self.hook.cancel_job(job_id=self.job_id,
project_id=self.project_id, location=self.location)
+
+
+class GreatExpectationsBigQueryOperator(BaseOperator):
Review comment:
All for it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
