You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2021/10/13 13:39:50 UTC
[GitHub] [airflow] wojsamjan opened a new pull request #18945: Add dataproc metastore operators
wojsamjan opened a new pull request #18945:
URL: https://github.com/apache/airflow/pull/18945
Add support for Google Dataproc Metastore. Includes operators, hooks, example dags, tests and docs.
Co-authored-by: Wojciech Januszek <ja...@google.com>
Co-authored-by: MaksYermak <ma...@google.com>
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in [UPDATING.md](https://github.com/apache/airflow/blob/main/UPDATING.md).
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mik-laj commented on a change in pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#discussion_r728296630
##########
File path: airflow/providers/google/cloud/example_dags/example_dataproc_metastore.py
##########
@@ -0,0 +1,202 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example Airflow DAG that show how to use various Dataproc Metastore
+operators to manage a service.
+"""
+
+import os
+
+from airflow import models
+from airflow.providers.google.cloud.operators.dataproc_metastore import (
+ DataprocMetastoreCreateBackupOperator,
+ DataprocMetastoreCreateMetadataImportOperator,
+ DataprocMetastoreCreateServiceOperator,
+ DataprocMetastoreDeleteBackupOperator,
+ DataprocMetastoreDeleteServiceOperator,
+ DataprocMetastoreExportMetadataOperator,
+ DataprocMetastoreGetServiceOperator,
+ DataprocMetastoreListBackupsOperator,
+ DataprocMetastoreRestoreServiceOperator,
+ DataprocMetastoreUpdateServiceOperator,
+)
+from airflow.utils.dates import days_ago
+
+PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "<PROJECT_ID>")
+SERVICE_ID = os.environ.get("GCP_DATAPROC_METASTORE_SERVICE_ID", "dataproc-metastore-system-tests-service-1")
+BACKUP_ID = os.environ.get("GCP_DATAPROC_METASTORE_BACKUP_ID", "dataproc-metastore-system-tests-backup-1")
+REGION = os.environ.get("GCP_REGION", "<REGION>")
+BUCKET = os.environ.get("GCP_DATAPROC_METASTORE_BUCKET", "dataproc-metastore-system-tests")
+METADATA_IMPORT_FILE = os.environ.get("GCS_METADATA_IMPORT_FILE", None)
+METADATA_IMPORT_ID = "dataproc-metastore-system-tests-metadata-import-1"
+TIMEOUT = 1200
+GCS_URI = f"gs://{BUCKET}/{METADATA_IMPORT_FILE}"
+DB_TYPE = "MYSQL"
+DESTINATION_GCS_FOLDER = "gs://{BUCKET}>"
+
+# Service definition
+# Docs: https://cloud.google.com/dataproc-metastore/docs/reference/rest/v1/projects.locations.services#Service
+# [START how_to_cloud_dataproc_metastore_create_service]
+SERVICE = {
+ "name": "test-service",
+}
+# [END how_to_cloud_dataproc_metastore_create_service]
+
+# Update service
+# [START how_to_cloud_dataproc_metastore_update_service]
+SERVICE_TO_UPDATE = {
+ "labels": {
+ "mylocalmachine": "mylocalmachine",
+ "systemtest": "systemtest",
+ }
+}
+UPDATE_MASK = {"paths": ["labels"]}
+# [END how_to_cloud_dataproc_metastore_update_service]
+
+# Backup definition
+# [START how_to_cloud_dataproc_metastore_create_backup]
+BACKUP = {
+ "name": "test-backup",
+}
+# [END how_to_cloud_dataproc_metastore_create_backup]
+
+# Metadata import definition
+# [START how_to_cloud_dataproc_metastore_create_metadata_import]
+METADATA_IMPORT = {
+ "name": "test-metadata-import",
+ "database_dump": {
+ "gcs_uri": GCS_URI,
+ "database_type": DB_TYPE,
+ },
+}
+# [END how_to_cloud_dataproc_metastore_create_metadata_import]
+
+
+with models.DAG("example_gcp_dataproc_metastore", start_date=days_ago(1), schedule_interval="@once") as dag:
+ # [START how_to_cloud_dataproc_metastore_create_service_operator]
+ create_service = DataprocMetastoreCreateServiceOperator(
+ task_id="create_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service=SERVICE,
+ service_id=SERVICE_ID,
+ timeout=TIMEOUT,
+ )
+ # [END how_to_cloud_dataproc_metastore_create_service_operator]
+
+ # [START how_to_cloud_dataproc_metastore_get_service_operator]
+ get_service_details = DataprocMetastoreGetServiceOperator(
+ task_id="get_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service_id=SERVICE_ID,
Review comment:
@josh-fell The example DAG files in the `airflow/providers/google` directory are automatically tested so we are sure they are working correctly. If we use blocks of code, it will be difficult for us to provide a similar level of confidence and trust. See: https://github.com/apache/airflow/blob/c1982eecad75c68e52a76088eb90ce15b9326436/tests/providers/google/cloud/operators/test_dataproc_metastore_system.py
Use `default_args` makes sense in the examples DAG in the core - [`airflow/example_dags`](https://github.com/apache/airflow/tree/main/airflow/example_dags). The examples in `airflow/providers` should first of all show how to use operators, not fancy features.
WDYT?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] wojsamjan commented on a change in pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
wojsamjan commented on a change in pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#discussion_r728871371
##########
File path: airflow/providers/google/cloud/hooks/dataproc_metastore.py
##########
@@ -0,0 +1,673 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+"""This module contains a Google Cloud Dataproc Metastore hook."""
+
+from time import sleep
+from typing import Dict, Optional, Sequence, Tuple, Union
+
+from google.api_core.operation import Operation
+from google.api_core.retry import Retry, exponential_sleep_generator
+from google.cloud.metastore_v1 import DataprocMetastoreClient
+from google.cloud.metastore_v1.types import Backup, MetadataImport, Service
+from google.cloud.metastore_v1.types.metastore import DatabaseDumpSpec, Restore
+from google.protobuf.field_mask_pb2 import FieldMask
+
+from airflow.exceptions import AirflowException
+from airflow.providers.google.common.hooks.base_google import GoogleBaseHook
+
+
+class DataprocMetastoreHook(GoogleBaseHook):
+ """Hook for Google Cloud Dataproc Metastore APIs."""
+
+ def get_dataproc_metastore_client(self, region: Optional[str] = None) -> DataprocMetastoreClient:
+ """Returns DataprocMetastoreClient."""
+ client_options = None
+ if region and region != 'global':
+ client_options = {'api_endpoint': f'{region}-metastore.googleapis.com:443'}
+
+ return DataprocMetastoreClient(
+ credentials=self._get_credentials(), client_info=self.client_info, client_options=client_options
+ )
+
+ def wait_for_operation(self, operation: Operation):
+ """Waits for long-lasting operation to complete."""
+ for time_to_wait in exponential_sleep_generator(initial=10, maximum=120):
+ sleep(time_to_wait)
+ if operation.done():
+ break
+
+ error = operation.exception()
+ if error:
+ raise AirflowException(error)
+ return operation.result()
+
+ def create_backup(
+ self,
+ project_id: str,
+ region: str,
+ service_id: str,
+ backup: Backup,
+ backup_id: str,
+ request_id: Optional[str] = None,
+ retry: Optional[Retry] = None,
+ timeout: Optional[float] = None,
+ metadata: Optional[Sequence[Tuple[str, str]]] = None,
+ ):
+ """
+ Creates a new backup in a given project and location.
+
+ :param project_id: Required. The ID of the Google Cloud project that the service belongs to.
+ :type project_id: str
+ :param region: Required. The ID of the Google Cloud region that the service belongs to.
+ :type region: str
+ :param service_id: Required. The ID of the metastore service, which is used as the final component of
+ the metastore service's name. This value must be between 2 and 63 characters long inclusive, begin
+ with a letter, end with a letter or number, and consist of alpha-numeric ASCII characters or
+ hyphens.
+
+ This corresponds to the ``service_id`` field on the ``request`` instance; if ``request`` is
+ provided, this should not be set.
+ :type service_id: str
+ :param backup: Required. The backup to create. The ``name`` field is ignored. The ID of the created
+ backup must be provided in the request's ``backup_id`` field.
+
+ This corresponds to the ``backup`` field on the ``request`` instance; if ``request`` is provided,
+ this should not be set.
+ :type backup: google.cloud.metastore_v1.types.Backup
+ :param backup_id: Required. The ID of the backup, which is used as the final component of the
+ backup's name. This value must be between 1 and 64 characters long, begin with a letter, end with
+ a letter or number, and consist of alpha-numeric ASCII characters or hyphens.
+
+ This corresponds to the ``backup_id`` field on the ``request`` instance; if ``request`` is
+ provided, this should not be set.
+ :type backup_id: str
+ :param request_id: Optional. A unique id used to identify the request.
+ :type request_id: str
+ :param retry: Designation of what errors, if any, should be retried.
+ :type retry: google.api_core.retry.Retry
+ :param timeout: The timeout for this request.
+ :type timeout: float
+ :param metadata: Strings which should be sent along with the request as metadata.
+ :type metadata: Sequence[Tuple[str, str]]
+ """
+ parent = f'projects/{project_id}/locations/{region}/services/{service_id}'
+
+ client = self.get_dataproc_metastore_client(region=region)
+ result = client.create_backup(
+ request={
+ 'parent': parent,
+ 'backup': backup,
+ 'backup_id': backup_id,
+ 'request_id': request_id,
+ },
+ retry=retry,
+ timeout=timeout,
+ metadata=metadata,
+ )
+ return result
+
+ def create_metadata_import(
Review comment:
Should I decorate another Hook´s methods for services and backup as well?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] lwyszomi commented on a change in pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
lwyszomi commented on a change in pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#discussion_r731683771
##########
File path: airflow/providers/google/cloud/hooks/dataproc_metastore.py
##########
@@ -0,0 +1,668 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+"""This module contains a Google Cloud Dataproc Metastore hook."""
+
+from typing import Dict, Optional, Sequence, Tuple, Union
+
+from google.api_core.operation import Operation
+from google.api_core.retry import Retry
+from google.cloud.metastore_v1 import DataprocMetastoreClient
+from google.cloud.metastore_v1.types import Backup, MetadataImport, Service
+from google.cloud.metastore_v1.types.metastore import DatabaseDumpSpec, Restore
+from google.protobuf.field_mask_pb2 import FieldMask
+
+from airflow.exceptions import AirflowException
+from airflow.providers.google.common.hooks.base_google import GoogleBaseHook
+
+
+class DataprocMetastoreHook(GoogleBaseHook):
+ """Hook for Google Cloud Dataproc Metastore APIs."""
+
+ def get_dataproc_metastore_client(self, region: Optional[str] = None) -> DataprocMetastoreClient:
+ """Returns DataprocMetastoreClient."""
+ client_options = None
+ if region and region != 'global':
+ client_options = {'api_endpoint': f'{region}-metastore.googleapis.com:443'}
+
+ return DataprocMetastoreClient(
+ credentials=self._get_credentials(), client_info=self.client_info, client_options=client_options
+ )
+
+ def wait_for_operation(self, timeout: float, operation: Operation):
+ """Waits for long-lasting operation to complete."""
+ error = operation.exception(timeout=timeout)
Review comment:
@wojsamjan This is not clear what happens here (from logic perspective :D ) maybe to make code more readable we can use this sample:
```
try:
return operation.result(timeout=timeout)
except Exception:
error = operaion.exception()
raise AirflowException(error)
```
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] wojsamjan closed pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
wojsamjan closed pull request #18945:
URL: https://github.com/apache/airflow/pull/18945
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] lwyszomi commented on a change in pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
lwyszomi commented on a change in pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#discussion_r731683771
##########
File path: airflow/providers/google/cloud/hooks/dataproc_metastore.py
##########
@@ -0,0 +1,668 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+"""This module contains a Google Cloud Dataproc Metastore hook."""
+
+from typing import Dict, Optional, Sequence, Tuple, Union
+
+from google.api_core.operation import Operation
+from google.api_core.retry import Retry
+from google.cloud.metastore_v1 import DataprocMetastoreClient
+from google.cloud.metastore_v1.types import Backup, MetadataImport, Service
+from google.cloud.metastore_v1.types.metastore import DatabaseDumpSpec, Restore
+from google.protobuf.field_mask_pb2 import FieldMask
+
+from airflow.exceptions import AirflowException
+from airflow.providers.google.common.hooks.base_google import GoogleBaseHook
+
+
+class DataprocMetastoreHook(GoogleBaseHook):
+ """Hook for Google Cloud Dataproc Metastore APIs."""
+
+ def get_dataproc_metastore_client(self, region: Optional[str] = None) -> DataprocMetastoreClient:
+ """Returns DataprocMetastoreClient."""
+ client_options = None
+ if region and region != 'global':
+ client_options = {'api_endpoint': f'{region}-metastore.googleapis.com:443'}
+
+ return DataprocMetastoreClient(
+ credentials=self._get_credentials(), client_info=self.client_info, client_options=client_options
+ )
+
+ def wait_for_operation(self, timeout: float, operation: Operation):
+ """Waits for long-lasting operation to complete."""
+ error = operation.exception(timeout=timeout)
Review comment:
This is not clear what happens here (from logic perspective :D ) maybe to make code more readable we can use this sample:
```
try:
return operation.result(timeout=timeout)
except Exception:
error = operaion.exception()
raise AirflowException(error)
```
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mik-laj commented on a change in pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#discussion_r728417941
##########
File path: airflow/providers/google/cloud/example_dags/example_dataproc_metastore.py
##########
@@ -0,0 +1,202 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example Airflow DAG that show how to use various Dataproc Metastore
+operators to manage a service.
+"""
+
+import os
+
+from airflow import models
+from airflow.providers.google.cloud.operators.dataproc_metastore import (
+ DataprocMetastoreCreateBackupOperator,
+ DataprocMetastoreCreateMetadataImportOperator,
+ DataprocMetastoreCreateServiceOperator,
+ DataprocMetastoreDeleteBackupOperator,
+ DataprocMetastoreDeleteServiceOperator,
+ DataprocMetastoreExportMetadataOperator,
+ DataprocMetastoreGetServiceOperator,
+ DataprocMetastoreListBackupsOperator,
+ DataprocMetastoreRestoreServiceOperator,
+ DataprocMetastoreUpdateServiceOperator,
+)
+from airflow.utils.dates import days_ago
+
+PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "<PROJECT_ID>")
+SERVICE_ID = os.environ.get("GCP_DATAPROC_METASTORE_SERVICE_ID", "dataproc-metastore-system-tests-service-1")
+BACKUP_ID = os.environ.get("GCP_DATAPROC_METASTORE_BACKUP_ID", "dataproc-metastore-system-tests-backup-1")
+REGION = os.environ.get("GCP_REGION", "<REGION>")
+BUCKET = os.environ.get("GCP_DATAPROC_METASTORE_BUCKET", "dataproc-metastore-system-tests")
+METADATA_IMPORT_FILE = os.environ.get("GCS_METADATA_IMPORT_FILE", None)
+METADATA_IMPORT_ID = "dataproc-metastore-system-tests-metadata-import-1"
+TIMEOUT = 1200
+GCS_URI = f"gs://{BUCKET}/{METADATA_IMPORT_FILE}"
+DB_TYPE = "MYSQL"
+DESTINATION_GCS_FOLDER = "gs://{BUCKET}>"
+
+# Service definition
+# Docs: https://cloud.google.com/dataproc-metastore/docs/reference/rest/v1/projects.locations.services#Service
+# [START how_to_cloud_dataproc_metastore_create_service]
+SERVICE = {
+ "name": "test-service",
+}
+# [END how_to_cloud_dataproc_metastore_create_service]
+
+# Update service
+# [START how_to_cloud_dataproc_metastore_update_service]
+SERVICE_TO_UPDATE = {
+ "labels": {
+ "mylocalmachine": "mylocalmachine",
+ "systemtest": "systemtest",
+ }
+}
+UPDATE_MASK = {"paths": ["labels"]}
+# [END how_to_cloud_dataproc_metastore_update_service]
+
+# Backup definition
+# [START how_to_cloud_dataproc_metastore_create_backup]
+BACKUP = {
+ "name": "test-backup",
+}
+# [END how_to_cloud_dataproc_metastore_create_backup]
+
+# Metadata import definition
+# [START how_to_cloud_dataproc_metastore_create_metadata_import]
+METADATA_IMPORT = {
+ "name": "test-metadata-import",
+ "database_dump": {
+ "gcs_uri": GCS_URI,
+ "database_type": DB_TYPE,
+ },
+}
+# [END how_to_cloud_dataproc_metastore_create_metadata_import]
+
+
+with models.DAG("example_gcp_dataproc_metastore", start_date=days_ago(1), schedule_interval="@once") as dag:
+ # [START how_to_cloud_dataproc_metastore_create_service_operator]
+ create_service = DataprocMetastoreCreateServiceOperator(
+ task_id="create_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service=SERVICE,
+ service_id=SERVICE_ID,
+ timeout=TIMEOUT,
+ )
+ # [END how_to_cloud_dataproc_metastore_create_service_operator]
+
+ # [START how_to_cloud_dataproc_metastore_get_service_operator]
+ get_service_details = DataprocMetastoreGetServiceOperator(
+ task_id="get_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service_id=SERVICE_ID,
Review comment:
This will work as long as the job definition is in a separate Python file. To add examples to documentation, we use `example-include` directive, To allow us to keep docs and example dags in sync
https://github.com/apache/airflow/blob/af4a5e006e4f5c9f203afeac039b22c6adee317f/docs/apache-airflow-providers-google/operators/cloud/dataproc.rst#L54-L58
Alternatively, we can also use the `code-block` directive, but this means that the example will not be tested and executed on CI. If it is not running on a CI, the examples very quickly become out-of-date and are more of a burden to maintain.
https://github.com/apache/airflow/blob/060345c0d982765e39da5fa8b2e2c6a01e89e394/docs/apache-airflow/tutorial.rst#L205-L231
For example:
In [this PR](https://github.com/apache/airflow/pull/18036/files), I deprecated default value of `pod_name` parameter in `EKSCreateClusterOperator`. This meant that if someone loads this file `DeprecationWarning` will be displayed. But we also [detect such problems](https://github.com/apache/airflow/pull/17900) automatically on CI, and we won't allow a merge PR that doesn't update the examples. This makes these examples much easier to maintain. We don't have to worry if all the examples have been updated to show the best practices because we detect all such problems automatically and if something still needs to be changed, we will inform everyone on the CI. If we used the `code-block` directive, we would have to find a way to detect similar problems automatically as well.
This is one example, but we have other automated checks that allow us to take care of the high quality of documentation and examples
> Just thought it was an opportunity for some legitimate use around default_args and document it.
I agree that this is a good use for `default_args`, but given the problems with using these examples in the documentation, we don't necessarily have to do.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] josh-fell commented on a change in pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
josh-fell commented on a change in pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#discussion_r728464727
##########
File path: airflow/providers/google/cloud/example_dags/example_dataproc_metastore.py
##########
@@ -0,0 +1,202 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example Airflow DAG that show how to use various Dataproc Metastore
+operators to manage a service.
+"""
+
+import os
+
+from airflow import models
+from airflow.providers.google.cloud.operators.dataproc_metastore import (
+ DataprocMetastoreCreateBackupOperator,
+ DataprocMetastoreCreateMetadataImportOperator,
+ DataprocMetastoreCreateServiceOperator,
+ DataprocMetastoreDeleteBackupOperator,
+ DataprocMetastoreDeleteServiceOperator,
+ DataprocMetastoreExportMetadataOperator,
+ DataprocMetastoreGetServiceOperator,
+ DataprocMetastoreListBackupsOperator,
+ DataprocMetastoreRestoreServiceOperator,
+ DataprocMetastoreUpdateServiceOperator,
+)
+from airflow.utils.dates import days_ago
+
+PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "<PROJECT_ID>")
+SERVICE_ID = os.environ.get("GCP_DATAPROC_METASTORE_SERVICE_ID", "dataproc-metastore-system-tests-service-1")
+BACKUP_ID = os.environ.get("GCP_DATAPROC_METASTORE_BACKUP_ID", "dataproc-metastore-system-tests-backup-1")
+REGION = os.environ.get("GCP_REGION", "<REGION>")
+BUCKET = os.environ.get("GCP_DATAPROC_METASTORE_BUCKET", "dataproc-metastore-system-tests")
+METADATA_IMPORT_FILE = os.environ.get("GCS_METADATA_IMPORT_FILE", None)
+METADATA_IMPORT_ID = "dataproc-metastore-system-tests-metadata-import-1"
+TIMEOUT = 1200
+GCS_URI = f"gs://{BUCKET}/{METADATA_IMPORT_FILE}"
+DB_TYPE = "MYSQL"
+DESTINATION_GCS_FOLDER = "gs://{BUCKET}>"
+
+# Service definition
+# Docs: https://cloud.google.com/dataproc-metastore/docs/reference/rest/v1/projects.locations.services#Service
+# [START how_to_cloud_dataproc_metastore_create_service]
+SERVICE = {
+ "name": "test-service",
+}
+# [END how_to_cloud_dataproc_metastore_create_service]
+
+# Update service
+# [START how_to_cloud_dataproc_metastore_update_service]
+SERVICE_TO_UPDATE = {
+ "labels": {
+ "mylocalmachine": "mylocalmachine",
+ "systemtest": "systemtest",
+ }
+}
+UPDATE_MASK = {"paths": ["labels"]}
+# [END how_to_cloud_dataproc_metastore_update_service]
+
+# Backup definition
+# [START how_to_cloud_dataproc_metastore_create_backup]
+BACKUP = {
+ "name": "test-backup",
+}
+# [END how_to_cloud_dataproc_metastore_create_backup]
+
+# Metadata import definition
+# [START how_to_cloud_dataproc_metastore_create_metadata_import]
+METADATA_IMPORT = {
+ "name": "test-metadata-import",
+ "database_dump": {
+ "gcs_uri": GCS_URI,
+ "database_type": DB_TYPE,
+ },
+}
+# [END how_to_cloud_dataproc_metastore_create_metadata_import]
+
+
+with models.DAG("example_gcp_dataproc_metastore", start_date=days_ago(1), schedule_interval="@once") as dag:
+ # [START how_to_cloud_dataproc_metastore_create_service_operator]
+ create_service = DataprocMetastoreCreateServiceOperator(
+ task_id="create_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service=SERVICE,
+ service_id=SERVICE_ID,
+ timeout=TIMEOUT,
+ )
+ # [END how_to_cloud_dataproc_metastore_create_service_operator]
+
+ # [START how_to_cloud_dataproc_metastore_get_service_operator]
+ get_service_details = DataprocMetastoreGetServiceOperator(
+ task_id="get_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service_id=SERVICE_ID,
Review comment:
@mik-laj Thanks for the context and explanation!
@wojsamjan Please ignore this `default_args` comment given the details explained above.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mik-laj commented on a change in pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#discussion_r728417941
##########
File path: airflow/providers/google/cloud/example_dags/example_dataproc_metastore.py
##########
@@ -0,0 +1,202 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example Airflow DAG that show how to use various Dataproc Metastore
+operators to manage a service.
+"""
+
+import os
+
+from airflow import models
+from airflow.providers.google.cloud.operators.dataproc_metastore import (
+ DataprocMetastoreCreateBackupOperator,
+ DataprocMetastoreCreateMetadataImportOperator,
+ DataprocMetastoreCreateServiceOperator,
+ DataprocMetastoreDeleteBackupOperator,
+ DataprocMetastoreDeleteServiceOperator,
+ DataprocMetastoreExportMetadataOperator,
+ DataprocMetastoreGetServiceOperator,
+ DataprocMetastoreListBackupsOperator,
+ DataprocMetastoreRestoreServiceOperator,
+ DataprocMetastoreUpdateServiceOperator,
+)
+from airflow.utils.dates import days_ago
+
+PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "<PROJECT_ID>")
+SERVICE_ID = os.environ.get("GCP_DATAPROC_METASTORE_SERVICE_ID", "dataproc-metastore-system-tests-service-1")
+BACKUP_ID = os.environ.get("GCP_DATAPROC_METASTORE_BACKUP_ID", "dataproc-metastore-system-tests-backup-1")
+REGION = os.environ.get("GCP_REGION", "<REGION>")
+BUCKET = os.environ.get("GCP_DATAPROC_METASTORE_BUCKET", "dataproc-metastore-system-tests")
+METADATA_IMPORT_FILE = os.environ.get("GCS_METADATA_IMPORT_FILE", None)
+METADATA_IMPORT_ID = "dataproc-metastore-system-tests-metadata-import-1"
+TIMEOUT = 1200
+GCS_URI = f"gs://{BUCKET}/{METADATA_IMPORT_FILE}"
+DB_TYPE = "MYSQL"
+DESTINATION_GCS_FOLDER = "gs://{BUCKET}>"
+
+# Service definition
+# Docs: https://cloud.google.com/dataproc-metastore/docs/reference/rest/v1/projects.locations.services#Service
+# [START how_to_cloud_dataproc_metastore_create_service]
+SERVICE = {
+ "name": "test-service",
+}
+# [END how_to_cloud_dataproc_metastore_create_service]
+
+# Update service
+# [START how_to_cloud_dataproc_metastore_update_service]
+SERVICE_TO_UPDATE = {
+ "labels": {
+ "mylocalmachine": "mylocalmachine",
+ "systemtest": "systemtest",
+ }
+}
+UPDATE_MASK = {"paths": ["labels"]}
+# [END how_to_cloud_dataproc_metastore_update_service]
+
+# Backup definition
+# [START how_to_cloud_dataproc_metastore_create_backup]
+BACKUP = {
+ "name": "test-backup",
+}
+# [END how_to_cloud_dataproc_metastore_create_backup]
+
+# Metadata import definition
+# [START how_to_cloud_dataproc_metastore_create_metadata_import]
+METADATA_IMPORT = {
+ "name": "test-metadata-import",
+ "database_dump": {
+ "gcs_uri": GCS_URI,
+ "database_type": DB_TYPE,
+ },
+}
+# [END how_to_cloud_dataproc_metastore_create_metadata_import]
+
+
+with models.DAG("example_gcp_dataproc_metastore", start_date=days_ago(1), schedule_interval="@once") as dag:
+ # [START how_to_cloud_dataproc_metastore_create_service_operator]
+ create_service = DataprocMetastoreCreateServiceOperator(
+ task_id="create_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service=SERVICE,
+ service_id=SERVICE_ID,
+ timeout=TIMEOUT,
+ )
+ # [END how_to_cloud_dataproc_metastore_create_service_operator]
+
+ # [START how_to_cloud_dataproc_metastore_get_service_operator]
+ get_service_details = DataprocMetastoreGetServiceOperator(
+ task_id="get_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service_id=SERVICE_ID,
Review comment:
This will work as long as the job definition is in a separate Python file. To add examples to documentation, we use `example-include` directive, To allow us to keep docs and example dags in sync
https://github.com/apache/airflow/blob/af4a5e006e4f5c9f203afeac039b22c6adee317f/docs/apache-airflow-providers-google/operators/cloud/dataproc.rst#L54-L58
Alternatively, we can also use the `code-block` directive, but this means that the example will not be tested and executed on CI. If it is not running on a CI, the examples very quickly become out-of-date and are more of a burden to maintain.
https://github.com/apache/airflow/blob/060345c0d982765e39da5fa8b2e2c6a01e89e394/docs/apache-airflow/tutorial.rst#L205-L231
For example:
In [this PR](https://github.com/apache/airflow/pull/18036/files), I deprecated default value of `pod_name` parameter in `EKSCreateClusterOperator`. This meant that if someone loads the original example file, `DeprecationWarning` will be displayed. But we also [detect such problems](https://github.com/apache/airflow/pull/17900) automatically on CI, and we won't allow a merge PR that doesn't update the examples. This makes these examples much easier to maintain. We don't have to worry if all the examples have been updated to show the best practices because we detect all such problems automatically and if something still needs to be changed, we will inform everyone on the CI. All this without the involvement of any of the project maintainers.
If we used the `code-block` directive, we would have to find a way to detect similar problems automatically as well.
This is one example, but we have other automated checks that allow us to take care of the high quality of documentation and examples
> Just thought it was an opportunity for some legitimate use around default_args and document it.
I agree that this is a good use for `default_args`, but given the problems with using these examples in the documentation, we don't necessarily have to do.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mik-laj commented on a change in pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#discussion_r728417941
##########
File path: airflow/providers/google/cloud/example_dags/example_dataproc_metastore.py
##########
@@ -0,0 +1,202 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example Airflow DAG that show how to use various Dataproc Metastore
+operators to manage a service.
+"""
+
+import os
+
+from airflow import models
+from airflow.providers.google.cloud.operators.dataproc_metastore import (
+ DataprocMetastoreCreateBackupOperator,
+ DataprocMetastoreCreateMetadataImportOperator,
+ DataprocMetastoreCreateServiceOperator,
+ DataprocMetastoreDeleteBackupOperator,
+ DataprocMetastoreDeleteServiceOperator,
+ DataprocMetastoreExportMetadataOperator,
+ DataprocMetastoreGetServiceOperator,
+ DataprocMetastoreListBackupsOperator,
+ DataprocMetastoreRestoreServiceOperator,
+ DataprocMetastoreUpdateServiceOperator,
+)
+from airflow.utils.dates import days_ago
+
+PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "<PROJECT_ID>")
+SERVICE_ID = os.environ.get("GCP_DATAPROC_METASTORE_SERVICE_ID", "dataproc-metastore-system-tests-service-1")
+BACKUP_ID = os.environ.get("GCP_DATAPROC_METASTORE_BACKUP_ID", "dataproc-metastore-system-tests-backup-1")
+REGION = os.environ.get("GCP_REGION", "<REGION>")
+BUCKET = os.environ.get("GCP_DATAPROC_METASTORE_BUCKET", "dataproc-metastore-system-tests")
+METADATA_IMPORT_FILE = os.environ.get("GCS_METADATA_IMPORT_FILE", None)
+METADATA_IMPORT_ID = "dataproc-metastore-system-tests-metadata-import-1"
+TIMEOUT = 1200
+GCS_URI = f"gs://{BUCKET}/{METADATA_IMPORT_FILE}"
+DB_TYPE = "MYSQL"
+DESTINATION_GCS_FOLDER = "gs://{BUCKET}>"
+
+# Service definition
+# Docs: https://cloud.google.com/dataproc-metastore/docs/reference/rest/v1/projects.locations.services#Service
+# [START how_to_cloud_dataproc_metastore_create_service]
+SERVICE = {
+ "name": "test-service",
+}
+# [END how_to_cloud_dataproc_metastore_create_service]
+
+# Update service
+# [START how_to_cloud_dataproc_metastore_update_service]
+SERVICE_TO_UPDATE = {
+ "labels": {
+ "mylocalmachine": "mylocalmachine",
+ "systemtest": "systemtest",
+ }
+}
+UPDATE_MASK = {"paths": ["labels"]}
+# [END how_to_cloud_dataproc_metastore_update_service]
+
+# Backup definition
+# [START how_to_cloud_dataproc_metastore_create_backup]
+BACKUP = {
+ "name": "test-backup",
+}
+# [END how_to_cloud_dataproc_metastore_create_backup]
+
+# Metadata import definition
+# [START how_to_cloud_dataproc_metastore_create_metadata_import]
+METADATA_IMPORT = {
+ "name": "test-metadata-import",
+ "database_dump": {
+ "gcs_uri": GCS_URI,
+ "database_type": DB_TYPE,
+ },
+}
+# [END how_to_cloud_dataproc_metastore_create_metadata_import]
+
+
+with models.DAG("example_gcp_dataproc_metastore", start_date=days_ago(1), schedule_interval="@once") as dag:
+ # [START how_to_cloud_dataproc_metastore_create_service_operator]
+ create_service = DataprocMetastoreCreateServiceOperator(
+ task_id="create_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service=SERVICE,
+ service_id=SERVICE_ID,
+ timeout=TIMEOUT,
+ )
+ # [END how_to_cloud_dataproc_metastore_create_service_operator]
+
+ # [START how_to_cloud_dataproc_metastore_get_service_operator]
+ get_service_details = DataprocMetastoreGetServiceOperator(
+ task_id="get_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service_id=SERVICE_ID,
Review comment:
This will work as long as the job definition is in a separate Python file. To add examples to documentation, we use `example-include` directive, To allow us to keep docs and example dags in sync
https://github.com/apache/airflow/blob/af4a5e006e4f5c9f203afeac039b22c6adee317f/docs/apache-airflow-providers-google/operators/cloud/dataproc.rst#L54-L58
Alternatively, we can also use the `code-block` directive, but this means that the example will not be tested and executed on CI. If it is not running on a CI, the examples very quickly become out-of-date and are more of a burden to maintain.
https://github.com/apache/airflow/blob/060345c0d982765e39da5fa8b2e2c6a01e89e394/docs/apache-airflow/tutorial.rst#L205-L231
For example:
In [this PR](https://github.com/apache/airflow/pull/18036/files), I deprecated default value of `pod_name` parameter in `EKSCreateClusterOperator`. This meant that if someone loads the original example file, `DeprecationWarning` will be displayed. But we also [detect such problems](https://github.com/apache/airflow/pull/17900) automatically on CI, and we won't allow a merge PR that doesn't update the examples. This makes these examples much easier to maintain. We don't have to worry if all the examples have been updated to show the best practices because we detect all such problems automatically and if something still needs to be changed, we will inform everyone on the CI. All this without the involvement of any of the project maintainers.
If we used the `code-block` directive, we would have to find a way to detect similar problems automatically as well, or the examples would be out of date very quickly.
This is one example, but we have other automated checks that allow us to take care of the high quality of documentation and examples
> Just thought it was an opportunity for some legitimate use around default_args and document it.
I agree that this is a good use for `default_args`, but given the problems with using these examples in the documentation, we don't necessarily have to do.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] lwyszomi commented on pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
lwyszomi commented on pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#issuecomment-958980408
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] lwyszomi commented on pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
lwyszomi commented on pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#issuecomment-949315221
@mik-laj Hey I fixed style errors and failing tests, now only failied `Tests / Quarantined tests (pull_request)` step, I think this is not related.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mik-laj commented on a change in pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#discussion_r728417941
##########
File path: airflow/providers/google/cloud/example_dags/example_dataproc_metastore.py
##########
@@ -0,0 +1,202 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example Airflow DAG that show how to use various Dataproc Metastore
+operators to manage a service.
+"""
+
+import os
+
+from airflow import models
+from airflow.providers.google.cloud.operators.dataproc_metastore import (
+ DataprocMetastoreCreateBackupOperator,
+ DataprocMetastoreCreateMetadataImportOperator,
+ DataprocMetastoreCreateServiceOperator,
+ DataprocMetastoreDeleteBackupOperator,
+ DataprocMetastoreDeleteServiceOperator,
+ DataprocMetastoreExportMetadataOperator,
+ DataprocMetastoreGetServiceOperator,
+ DataprocMetastoreListBackupsOperator,
+ DataprocMetastoreRestoreServiceOperator,
+ DataprocMetastoreUpdateServiceOperator,
+)
+from airflow.utils.dates import days_ago
+
+PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "<PROJECT_ID>")
+SERVICE_ID = os.environ.get("GCP_DATAPROC_METASTORE_SERVICE_ID", "dataproc-metastore-system-tests-service-1")
+BACKUP_ID = os.environ.get("GCP_DATAPROC_METASTORE_BACKUP_ID", "dataproc-metastore-system-tests-backup-1")
+REGION = os.environ.get("GCP_REGION", "<REGION>")
+BUCKET = os.environ.get("GCP_DATAPROC_METASTORE_BUCKET", "dataproc-metastore-system-tests")
+METADATA_IMPORT_FILE = os.environ.get("GCS_METADATA_IMPORT_FILE", None)
+METADATA_IMPORT_ID = "dataproc-metastore-system-tests-metadata-import-1"
+TIMEOUT = 1200
+GCS_URI = f"gs://{BUCKET}/{METADATA_IMPORT_FILE}"
+DB_TYPE = "MYSQL"
+DESTINATION_GCS_FOLDER = "gs://{BUCKET}>"
+
+# Service definition
+# Docs: https://cloud.google.com/dataproc-metastore/docs/reference/rest/v1/projects.locations.services#Service
+# [START how_to_cloud_dataproc_metastore_create_service]
+SERVICE = {
+ "name": "test-service",
+}
+# [END how_to_cloud_dataproc_metastore_create_service]
+
+# Update service
+# [START how_to_cloud_dataproc_metastore_update_service]
+SERVICE_TO_UPDATE = {
+ "labels": {
+ "mylocalmachine": "mylocalmachine",
+ "systemtest": "systemtest",
+ }
+}
+UPDATE_MASK = {"paths": ["labels"]}
+# [END how_to_cloud_dataproc_metastore_update_service]
+
+# Backup definition
+# [START how_to_cloud_dataproc_metastore_create_backup]
+BACKUP = {
+ "name": "test-backup",
+}
+# [END how_to_cloud_dataproc_metastore_create_backup]
+
+# Metadata import definition
+# [START how_to_cloud_dataproc_metastore_create_metadata_import]
+METADATA_IMPORT = {
+ "name": "test-metadata-import",
+ "database_dump": {
+ "gcs_uri": GCS_URI,
+ "database_type": DB_TYPE,
+ },
+}
+# [END how_to_cloud_dataproc_metastore_create_metadata_import]
+
+
+with models.DAG("example_gcp_dataproc_metastore", start_date=days_ago(1), schedule_interval="@once") as dag:
+ # [START how_to_cloud_dataproc_metastore_create_service_operator]
+ create_service = DataprocMetastoreCreateServiceOperator(
+ task_id="create_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service=SERVICE,
+ service_id=SERVICE_ID,
+ timeout=TIMEOUT,
+ )
+ # [END how_to_cloud_dataproc_metastore_create_service_operator]
+
+ # [START how_to_cloud_dataproc_metastore_get_service_operator]
+ get_service_details = DataprocMetastoreGetServiceOperator(
+ task_id="get_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service_id=SERVICE_ID,
Review comment:
This will work as long as the task definition is in a separate Python file. To add examples to documentation, we use `example-include` directive, To allow us to keep docs and example dags in sync
https://github.com/apache/airflow/blob/af4a5e006e4f5c9f203afeac039b22c6adee317f/docs/apache-airflow-providers-google/operators/cloud/dataproc.rst#L54-L58
Alternatively, we can also use the `code-block` directive, but this means that the example will not be tested and executed on CI. If it is not running on a CI, the examples very quickly become out-of-date and are more of a burden to maintain.
https://github.com/apache/airflow/blob/060345c0d982765e39da5fa8b2e2c6a01e89e394/docs/apache-airflow/tutorial.rst#L205-L231
For example:
In [this PR](https://github.com/apache/airflow/pull/18036/files), I deprecated default value of `pod_name` parameter in `EKSCreateClusterOperator`. This meant that if someone loads the original example file, `DeprecationWarning` will be displayed. But we also [detect such problems](https://github.com/apache/airflow/pull/17900) automatically on CI, and we won't allow a merge PR that doesn't update the examples. This makes these examples much easier to maintain. We don't have to worry if all the examples have been updated to show the best practices because we detect all such problems automatically and if something still needs to be changed, we will inform everyone on the CI. All this without the involvement of any of the project maintainers.
If we used the `code-block` directive, we would have to find a way to detect similar problems automatically as well, or the examples would be out of date very quickly.
This is one example, but we have other automated checks that allow us to take care of the high quality of documentation and examples
> Just thought it was an opportunity for some legitimate use around default_args and document it.
I agree that this is a good use for `default_args`, but given the problems with using these examples in the documentation, we don't necessarily have to do in all cases.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] wojsamjan commented on a change in pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
wojsamjan commented on a change in pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#discussion_r728871371
##########
File path: airflow/providers/google/cloud/hooks/dataproc_metastore.py
##########
@@ -0,0 +1,673 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+"""This module contains a Google Cloud Dataproc Metastore hook."""
+
+from time import sleep
+from typing import Dict, Optional, Sequence, Tuple, Union
+
+from google.api_core.operation import Operation
+from google.api_core.retry import Retry, exponential_sleep_generator
+from google.cloud.metastore_v1 import DataprocMetastoreClient
+from google.cloud.metastore_v1.types import Backup, MetadataImport, Service
+from google.cloud.metastore_v1.types.metastore import DatabaseDumpSpec, Restore
+from google.protobuf.field_mask_pb2 import FieldMask
+
+from airflow.exceptions import AirflowException
+from airflow.providers.google.common.hooks.base_google import GoogleBaseHook
+
+
+class DataprocMetastoreHook(GoogleBaseHook):
+ """Hook for Google Cloud Dataproc Metastore APIs."""
+
+ def get_dataproc_metastore_client(self, region: Optional[str] = None) -> DataprocMetastoreClient:
+ """Returns DataprocMetastoreClient."""
+ client_options = None
+ if region and region != 'global':
+ client_options = {'api_endpoint': f'{region}-metastore.googleapis.com:443'}
+
+ return DataprocMetastoreClient(
+ credentials=self._get_credentials(), client_info=self.client_info, client_options=client_options
+ )
+
+ def wait_for_operation(self, operation: Operation):
+ """Waits for long-lasting operation to complete."""
+ for time_to_wait in exponential_sleep_generator(initial=10, maximum=120):
+ sleep(time_to_wait)
+ if operation.done():
+ break
+
+ error = operation.exception()
+ if error:
+ raise AirflowException(error)
+ return operation.result()
+
+ def create_backup(
+ self,
+ project_id: str,
+ region: str,
+ service_id: str,
+ backup: Backup,
+ backup_id: str,
+ request_id: Optional[str] = None,
+ retry: Optional[Retry] = None,
+ timeout: Optional[float] = None,
+ metadata: Optional[Sequence[Tuple[str, str]]] = None,
+ ):
+ """
+ Creates a new backup in a given project and location.
+
+ :param project_id: Required. The ID of the Google Cloud project that the service belongs to.
+ :type project_id: str
+ :param region: Required. The ID of the Google Cloud region that the service belongs to.
+ :type region: str
+ :param service_id: Required. The ID of the metastore service, which is used as the final component of
+ the metastore service's name. This value must be between 2 and 63 characters long inclusive, begin
+ with a letter, end with a letter or number, and consist of alpha-numeric ASCII characters or
+ hyphens.
+
+ This corresponds to the ``service_id`` field on the ``request`` instance; if ``request`` is
+ provided, this should not be set.
+ :type service_id: str
+ :param backup: Required. The backup to create. The ``name`` field is ignored. The ID of the created
+ backup must be provided in the request's ``backup_id`` field.
+
+ This corresponds to the ``backup`` field on the ``request`` instance; if ``request`` is provided,
+ this should not be set.
+ :type backup: google.cloud.metastore_v1.types.Backup
+ :param backup_id: Required. The ID of the backup, which is used as the final component of the
+ backup's name. This value must be between 1 and 64 characters long, begin with a letter, end with
+ a letter or number, and consist of alpha-numeric ASCII characters or hyphens.
+
+ This corresponds to the ``backup_id`` field on the ``request`` instance; if ``request`` is
+ provided, this should not be set.
+ :type backup_id: str
+ :param request_id: Optional. A unique id used to identify the request.
+ :type request_id: str
+ :param retry: Designation of what errors, if any, should be retried.
+ :type retry: google.api_core.retry.Retry
+ :param timeout: The timeout for this request.
+ :type timeout: float
+ :param metadata: Strings which should be sent along with the request as metadata.
+ :type metadata: Sequence[Tuple[str, str]]
+ """
+ parent = f'projects/{project_id}/locations/{region}/services/{service_id}'
+
+ client = self.get_dataproc_metastore_client(region=region)
+ result = client.create_backup(
+ request={
+ 'parent': parent,
+ 'backup': backup,
+ 'backup_id': backup_id,
+ 'request_id': request_id,
+ },
+ retry=retry,
+ timeout=timeout,
+ metadata=metadata,
+ )
+ return result
+
+ def create_metadata_import(
Review comment:
Should I decorate another Hook´s methods for services and backup as well? @mik-laj
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] wojsamjan commented on a change in pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
wojsamjan commented on a change in pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#discussion_r728882220
##########
File path: airflow/providers/google/cloud/hooks/dataproc_metastore.py
##########
@@ -0,0 +1,673 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+"""This module contains a Google Cloud Dataproc Metastore hook."""
+
+from time import sleep
+from typing import Dict, Optional, Sequence, Tuple, Union
+
+from google.api_core.operation import Operation
+from google.api_core.retry import Retry, exponential_sleep_generator
+from google.cloud.metastore_v1 import DataprocMetastoreClient
+from google.cloud.metastore_v1.types import Backup, MetadataImport, Service
+from google.cloud.metastore_v1.types.metastore import DatabaseDumpSpec, Restore
+from google.protobuf.field_mask_pb2 import FieldMask
+
+from airflow.exceptions import AirflowException
+from airflow.providers.google.common.hooks.base_google import GoogleBaseHook
+
+
+class DataprocMetastoreHook(GoogleBaseHook):
+ """Hook for Google Cloud Dataproc Metastore APIs."""
+
+ def get_dataproc_metastore_client(self, region: Optional[str] = None) -> DataprocMetastoreClient:
+ """Returns DataprocMetastoreClient."""
+ client_options = None
+ if region and region != 'global':
+ client_options = {'api_endpoint': f'{region}-metastore.googleapis.com:443'}
+
+ return DataprocMetastoreClient(
+ credentials=self._get_credentials(), client_info=self.client_info, client_options=client_options
+ )
+
+ def wait_for_operation(self, operation: Operation):
+ """Waits for long-lasting operation to complete."""
+ for time_to_wait in exponential_sleep_generator(initial=10, maximum=120):
+ sleep(time_to_wait)
+ if operation.done():
+ break
Review comment:
@mik-laj As I understand after the change I will have to pass timeout every time to the Hook´s wait_for_operation so both https://github.com/googleapis/python-api-core/blob/main/google/api_core/future/polling.py#L116 & https://github.com/googleapis/python-api-core/blob/main/google/api_core/future/polling.py#L141 could call https://github.com/googleapis/python-api-core/blob/main/google/api_core/future/polling.py#L106 with timeout passed for Retry´s deadline, right?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mik-laj commented on pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
mik-laj commented on pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#issuecomment-948179409
These changes looks related to this contribution
```
==== Providers postgres: 13 failures ====
tests/providers/google/cloud/hooks/test_dataproc_metastore.py::TestDataprocMetastoreWithDefaultProjectIdHook::test_create_service: AssertionError: Expected call: create_service(metadata=None, request={'parent': 'projects/test-project-id/locations/test-region', 'service_id': 'test-service-id', 'service': {}, 'request_id': None}, retry=None, timeout=None)
Actual call: create_service(metadata=(), request={'parent': 'projects/test-project-id/locations/test-region', 'service_id': 'test-service-id', 'service': {'name': 'test-service'}, 'request_id': None}, retry=None, timeout=None)
tests/providers/google/cloud/hooks/test_dataproc_metastore.py::TestDataprocMetastoreWithDefaultProjectIdHook::test_update_service: google.auth.exceptions.DefaultCredentialsError: Could not automatically determine credentials. Please set GOOGLE_APPLICATION_CREDENTIALS or explicitly create credentials and re-run the application. For more information, please see https://cloud.google.com/docs/authentication/getting-started
tests/providers/google/cloud/hooks/test_dataproc_metastore.py::TestDataprocMetastoreWithoutDefaultProjectIdHook::test_create_service: AssertionError: Expected call: create_service(metadata=None, request={'parent': 'projects/test-project-id/locations/test-region', 'service_id': 'test-service-id', 'service': {'name': 'test-service'}, 'request_id': None}, retry=None, timeout=None)
Actual call: create_service(metadata=(), request={'parent': 'projects/test-project-id/locations/test-region', 'service_id': 'test-service-id', 'service': {'name': 'test-service'}, 'request_id': None}, retry=None, timeout=None)
tests/providers/google/cloud/hooks/test_dataproc_metastore.py::TestDataprocMetastoreWithoutDefaultProjectIdHook::test_update_service: google.auth.exceptions.DefaultCredentialsError: Could not automatically determine credentials. Please set GOOGLE_APPLICATION_CREDENTIALS or explicitly create credentials and re-run the application. For more information, please see https://cloud.google.com/docs/authentication/getting-started
tests/providers/google/cloud/operators/test_dataproc_metastore.py::TestDataprocMetastoreCreateBackupOperator::test_assert_valid_hook_call: TypeError: <MagicMock name='DataprocMetastoreHook().wait_for_operation()' id='140010347401960'> is not an instance of Backup
tests/providers/google/cloud/operators/test_dataproc_metastore.py::TestDataprocMetastoreCreateMetadataImportOperator::test_assert_valid_hook_call: TypeError: <MagicMock name='DataprocMetastoreHook().wait_for_operation()' id='140010430133976'> is not an instance of MetadataImport
tests/providers/google/cloud/operators/test_dataproc_metastore.py::TestDataprocMetastoreCreateServiceOperator::test_execute: TypeError: <MagicMock name='DataprocMetastoreHook().wait_for_operation()' id='140010332321160'> is not an instance of Service
tests/providers/google/cloud/operators/test_dataproc_metastore.py::TestDataprocMetastoreDeleteBackupOperator::test_assert_valid_hook_call: AssertionError: Expected call: DataprocMetastoreHook(gcp_conn_id='test-gcp-conn-id')
Actual call: DataprocMetastoreHook(gcp_conn_id='test-gcp-conn-id', impersonation_chain=None)
tests/providers/google/cloud/operators/test_dataproc_metastore.py::TestDataprocMetastoreExportMetadataOperator::test_assert_valid_hook_call: airflow.exceptions.AirflowException: Argument ['service_id'] is required
tests/providers/google/cloud/operators/test_dataproc_metastore.py::TestDataprocMetastoreGetServiceOperator::test_execute: TypeError: <MagicMock name='DataprocMetastoreHook().get_service()' id='140010395853488'> is not an instance of Service
tests/providers/google/cloud/operators/test_dataproc_metastore.py::TestDataprocMetastoreListBackupsOperator::test_assert_valid_hook_call: AssertionError: Expected call: DataprocMetastoreHook(gcp_conn_id='test-gcp-conn-id')
Actual call: DataprocMetastoreHook(gcp_conn_id='test-gcp-conn-id', impersonation_chain=None)
tests/providers/google/cloud/operators/test_dataproc_metastore.py::TestDataprocMetastoreRestoreServiceOperator::test_assert_valid_hook_call: Failed: Timeout >60.0s
tests/providers/google/cloud/operators/test_dataproc_metastore.py::TestDataprocMetastoreUpdateServiceOperator::test_assert_valid_hook_call: AssertionError: Expected call: DataprocMetastoreHook(gcp_conn_id='test-gcp-conn-id')
Actual call: DataprocMetastoreHook(gcp_conn_id='test-gcp-conn-id', impersonation_chain=None)
Error: Process completed with exit code 1.
```
https://github.com/apache/airflow/pull/18945/checks?check_run_id=3953505880
```
Run codespell to check for common misspellings in files..................................Failed
- hook id: codespell
- exit code: 65
airflow/providers/google/cloud/operators/dataproc_metastore.py:46: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/operators/dataproc_metastore.py:60: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/operators/dataproc_metastore.py:168: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/operators/dataproc_metastore.py:182: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/operators/dataproc_metastore.py:281: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/operators/dataproc_metastore.py:385: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/operators/dataproc_metastore.py:393: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/operators/dataproc_metastore.py:549: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/operators/dataproc_metastore.py:664: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/operators/dataproc_metastore.py:743: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/operators/dataproc_metastore.py:834: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/operators/dataproc_metastore.py:849: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/operators/dataproc_metastore.py:974: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/hooks/dataproc_metastore.py:76: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/hooks/dataproc_metastore.py:90: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/hooks/dataproc_metastore.py:142: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/hooks/dataproc_metastore.py:156: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/hooks/dataproc_metastore.py:214: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/hooks/dataproc_metastore.py:266: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/hooks/dataproc_metastore.py:274: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/hooks/dataproc_metastore.py:322: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/hooks/dataproc_metastore.py:378: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/hooks/dataproc_metastore.py:431: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/hooks/dataproc_metastore.py:480: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/hooks/dataproc_metastore.py:558: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/hooks/dataproc_metastore.py:573: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/hooks/dataproc_metastore.py:631: alpha-numeric ==> alphanumeric
```
https://github.com/apache/airflow/pull/18945/checks?check_run_id=3953505494
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] josh-fell commented on a change in pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
josh-fell commented on a change in pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#discussion_r728128151
##########
File path: airflow/providers/google/cloud/example_dags/example_dataproc_metastore.py
##########
@@ -0,0 +1,202 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example Airflow DAG that show how to use various Dataproc Metastore
+operators to manage a service.
+"""
+
+import os
+
+from airflow import models
+from airflow.providers.google.cloud.operators.dataproc_metastore import (
+ DataprocMetastoreCreateBackupOperator,
+ DataprocMetastoreCreateMetadataImportOperator,
+ DataprocMetastoreCreateServiceOperator,
+ DataprocMetastoreDeleteBackupOperator,
+ DataprocMetastoreDeleteServiceOperator,
+ DataprocMetastoreExportMetadataOperator,
+ DataprocMetastoreGetServiceOperator,
+ DataprocMetastoreListBackupsOperator,
+ DataprocMetastoreRestoreServiceOperator,
+ DataprocMetastoreUpdateServiceOperator,
+)
+from airflow.utils.dates import days_ago
+
+PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "<PROJECT_ID>")
+SERVICE_ID = os.environ.get("GCP_DATAPROC_METASTORE_SERVICE_ID", "dataproc-metastore-system-tests-service-1")
+BACKUP_ID = os.environ.get("GCP_DATAPROC_METASTORE_BACKUP_ID", "dataproc-metastore-system-tests-backup-1")
+REGION = os.environ.get("GCP_REGION", "<REGION>")
+BUCKET = os.environ.get("GCP_DATAPROC_METASTORE_BUCKET", "dataproc-metastore-system-tests")
+METADATA_IMPORT_FILE = os.environ.get("GCS_METADATA_IMPORT_FILE", None)
+METADATA_IMPORT_ID = "dataproc-metastore-system-tests-metadata-import-1"
+TIMEOUT = 1200
+GCS_URI = f"gs://{BUCKET}/{METADATA_IMPORT_FILE}"
+DB_TYPE = "MYSQL"
+DESTINATION_GCS_FOLDER = "gs://{BUCKET}>"
+
+# Service definition
+# Docs: https://cloud.google.com/dataproc-metastore/docs/reference/rest/v1/projects.locations.services#Service
+# [START how_to_cloud_dataproc_metastore_create_service]
+SERVICE = {
+ "name": "test-service",
+}
+# [END how_to_cloud_dataproc_metastore_create_service]
+
+# Update service
+# [START how_to_cloud_dataproc_metastore_update_service]
+SERVICE_TO_UPDATE = {
+ "labels": {
+ "mylocalmachine": "mylocalmachine",
+ "systemtest": "systemtest",
+ }
+}
+UPDATE_MASK = {"paths": ["labels"]}
+# [END how_to_cloud_dataproc_metastore_update_service]
+
+# Backup definition
+# [START how_to_cloud_dataproc_metastore_create_backup]
+BACKUP = {
+ "name": "test-backup",
+}
+# [END how_to_cloud_dataproc_metastore_create_backup]
+
+# Metadata import definition
+# [START how_to_cloud_dataproc_metastore_create_metadata_import]
+METADATA_IMPORT = {
+ "name": "test-metadata-import",
+ "database_dump": {
+ "gcs_uri": GCS_URI,
+ "database_type": DB_TYPE,
+ },
+}
+# [END how_to_cloud_dataproc_metastore_create_metadata_import]
+
+
+with models.DAG("example_gcp_dataproc_metastore", start_date=days_ago(1), schedule_interval="@once") as dag:
Review comment:
There is an ongoing effort to update the example DAGs in Airflow to transition away from `days_ago(n)` for `start_date` to a static value, improve `default_args` use, etc. when applicable. For new DAGs, let's follow suit for using a static `start_date` value as it is best practice.
##########
File path: airflow/providers/google/cloud/example_dags/example_dataproc_metastore.py
##########
@@ -0,0 +1,202 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example Airflow DAG that show how to use various Dataproc Metastore
+operators to manage a service.
+"""
+
+import os
+
+from airflow import models
+from airflow.providers.google.cloud.operators.dataproc_metastore import (
+ DataprocMetastoreCreateBackupOperator,
+ DataprocMetastoreCreateMetadataImportOperator,
+ DataprocMetastoreCreateServiceOperator,
+ DataprocMetastoreDeleteBackupOperator,
+ DataprocMetastoreDeleteServiceOperator,
+ DataprocMetastoreExportMetadataOperator,
+ DataprocMetastoreGetServiceOperator,
+ DataprocMetastoreListBackupsOperator,
+ DataprocMetastoreRestoreServiceOperator,
+ DataprocMetastoreUpdateServiceOperator,
+)
+from airflow.utils.dates import days_ago
+
+PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "<PROJECT_ID>")
+SERVICE_ID = os.environ.get("GCP_DATAPROC_METASTORE_SERVICE_ID", "dataproc-metastore-system-tests-service-1")
+BACKUP_ID = os.environ.get("GCP_DATAPROC_METASTORE_BACKUP_ID", "dataproc-metastore-system-tests-backup-1")
+REGION = os.environ.get("GCP_REGION", "<REGION>")
+BUCKET = os.environ.get("GCP_DATAPROC_METASTORE_BUCKET", "dataproc-metastore-system-tests")
+METADATA_IMPORT_FILE = os.environ.get("GCS_METADATA_IMPORT_FILE", None)
+METADATA_IMPORT_ID = "dataproc-metastore-system-tests-metadata-import-1"
+TIMEOUT = 1200
+GCS_URI = f"gs://{BUCKET}/{METADATA_IMPORT_FILE}"
+DB_TYPE = "MYSQL"
+DESTINATION_GCS_FOLDER = "gs://{BUCKET}>"
+
+# Service definition
+# Docs: https://cloud.google.com/dataproc-metastore/docs/reference/rest/v1/projects.locations.services#Service
+# [START how_to_cloud_dataproc_metastore_create_service]
+SERVICE = {
+ "name": "test-service",
+}
+# [END how_to_cloud_dataproc_metastore_create_service]
+
+# Update service
+# [START how_to_cloud_dataproc_metastore_update_service]
+SERVICE_TO_UPDATE = {
+ "labels": {
+ "mylocalmachine": "mylocalmachine",
+ "systemtest": "systemtest",
+ }
+}
+UPDATE_MASK = {"paths": ["labels"]}
+# [END how_to_cloud_dataproc_metastore_update_service]
+
+# Backup definition
+# [START how_to_cloud_dataproc_metastore_create_backup]
+BACKUP = {
+ "name": "test-backup",
+}
+# [END how_to_cloud_dataproc_metastore_create_backup]
+
+# Metadata import definition
+# [START how_to_cloud_dataproc_metastore_create_metadata_import]
+METADATA_IMPORT = {
+ "name": "test-metadata-import",
+ "database_dump": {
+ "gcs_uri": GCS_URI,
+ "database_type": DB_TYPE,
+ },
+}
+# [END how_to_cloud_dataproc_metastore_create_metadata_import]
+
+
+with models.DAG("example_gcp_dataproc_metastore", start_date=days_ago(1), schedule_interval="@once") as dag:
+ # [START how_to_cloud_dataproc_metastore_create_service_operator]
+ create_service = DataprocMetastoreCreateServiceOperator(
+ task_id="create_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service=SERVICE,
+ service_id=SERVICE_ID,
+ timeout=TIMEOUT,
+ )
+ # [END how_to_cloud_dataproc_metastore_create_service_operator]
+
+ # [START how_to_cloud_dataproc_metastore_get_service_operator]
+ get_service_details = DataprocMetastoreGetServiceOperator(
+ task_id="get_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service_id=SERVICE_ID,
+ )
+ # [END how_to_cloud_dataproc_metastore_get_service_operator]
+
+ # [START how_to_cloud_dataproc_metastore_update_service_operator]
+ update_service = DataprocMetastoreUpdateServiceOperator(
+ task_id="update_service",
+ project_id=PROJECT_ID,
+ service_id=SERVICE_ID,
+ region=REGION,
+ service=SERVICE_TO_UPDATE,
+ update_mask=UPDATE_MASK,
+ )
+ # [END how_to_cloud_dataproc_metastore_update_service_operator]
+
+ # [START how_to_cloud_dataproc_metastore_create_metadata_import_operator]
+ import_metadata = DataprocMetastoreCreateMetadataImportOperator(
+ task_id="create_metadata_import",
+ project_id=PROJECT_ID,
+ region=REGION,
+ service_id=SERVICE_ID,
+ metadata_import=METADATA_IMPORT,
+ metadata_import_id=METADATA_IMPORT_ID,
+ )
+ # [END how_to_cloud_dataproc_metastore_create_metadata_import_operator]
+
+ # [START how_to_cloud_dataproc_metastore_export_metadata_operator]
+ export_metadata = DataprocMetastoreExportMetadataOperator(
+ task_id="export_metadata",
+ destination_gcs_folder=DESTINATION_GCS_FOLDER,
+ project_id=PROJECT_ID,
+ region=REGION,
+ service_id=SERVICE_ID,
+ timeout=TIMEOUT,
+ )
+ # [END how_to_cloud_dataproc_metastore_export_metadata_operator]
+
+ # [START how_to_cloud_dataproc_metastore_create_backup_operator]
+ backup_service = DataprocMetastoreCreateBackupOperator(
+ task_id="create_backup",
+ project_id=PROJECT_ID,
+ region=REGION,
+ service_id=SERVICE_ID,
+ backup=BACKUP,
+ backup_id=BACKUP_ID,
+ )
+ # [END how_to_cloud_dataproc_metastore_create_backup_operator]
+
+ # [START how_to_cloud_dataproc_metastore_list_backups_operator]
+ list_backups = DataprocMetastoreListBackupsOperator(
+ task_id="list_backups",
+ project_id=PROJECT_ID,
+ region=REGION,
+ service_id=SERVICE_ID,
+ )
+ # [END how_to_cloud_dataproc_metastore_list_backups_operator]
+
+ # [START how_to_cloud_dataproc_metastore_delete_backup_operator]
+ delete_backup = DataprocMetastoreDeleteBackupOperator(
+ task_id="delete_backup",
+ project_id=PROJECT_ID,
+ region=REGION,
+ service_id=SERVICE_ID,
+ backup_id=BACKUP_ID,
+ )
+ # [END how_to_cloud_dataproc_metastore_delete_backup_operator]
+
+ # [START how_to_cloud_dataproc_metastore_restore_service_operator]
+ restore_service = DataprocMetastoreRestoreServiceOperator(
+ task_id="restore_metastore",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service_id=SERVICE_ID,
+ backup_id=BACKUP_ID,
+ backup_region=REGION,
+ backup_project_id=PROJECT_ID,
+ backup_service_id=SERVICE_ID,
+ )
+ # [END how_to_cloud_dataproc_metastore_restore_service_operator]
+
+ # [START how_to_cloud_dataproc_metastore_delete_service_operator]
+ delete_service = DataprocMetastoreDeleteServiceOperator(
+ task_id="delete_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service_id=SERVICE_ID,
+ )
+ # [END how_to_cloud_dataproc_metastore_delete_service_operator]
+
+ create_service >> update_service >> get_service_details
+
+ get_service_details >> backup_service >> list_backups >> restore_service >> delete_backup
+
+ delete_backup >> export_metadata >> import_metadata
+
+ import_metadata >> delete_service
Review comment:
WDYT about using `airflow.models.baseoperator.chain()` here? It may be more readable to do something like
```python
chain(
create_service,
update_service,
get_service_details,
backup_service,
list_backups,
restore_service,
delete_backup,
export_metadata,
import_metadata,
delete_service,
)
```
##########
File path: airflow/providers/google/cloud/example_dags/example_dataproc_metastore.py
##########
@@ -0,0 +1,202 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example Airflow DAG that show how to use various Dataproc Metastore
+operators to manage a service.
+"""
+
+import os
+
+from airflow import models
+from airflow.providers.google.cloud.operators.dataproc_metastore import (
+ DataprocMetastoreCreateBackupOperator,
+ DataprocMetastoreCreateMetadataImportOperator,
+ DataprocMetastoreCreateServiceOperator,
+ DataprocMetastoreDeleteBackupOperator,
+ DataprocMetastoreDeleteServiceOperator,
+ DataprocMetastoreExportMetadataOperator,
+ DataprocMetastoreGetServiceOperator,
+ DataprocMetastoreListBackupsOperator,
+ DataprocMetastoreRestoreServiceOperator,
+ DataprocMetastoreUpdateServiceOperator,
+)
+from airflow.utils.dates import days_ago
+
+PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "<PROJECT_ID>")
+SERVICE_ID = os.environ.get("GCP_DATAPROC_METASTORE_SERVICE_ID", "dataproc-metastore-system-tests-service-1")
+BACKUP_ID = os.environ.get("GCP_DATAPROC_METASTORE_BACKUP_ID", "dataproc-metastore-system-tests-backup-1")
+REGION = os.environ.get("GCP_REGION", "<REGION>")
+BUCKET = os.environ.get("GCP_DATAPROC_METASTORE_BUCKET", "dataproc-metastore-system-tests")
+METADATA_IMPORT_FILE = os.environ.get("GCS_METADATA_IMPORT_FILE", None)
+METADATA_IMPORT_ID = "dataproc-metastore-system-tests-metadata-import-1"
+TIMEOUT = 1200
+GCS_URI = f"gs://{BUCKET}/{METADATA_IMPORT_FILE}"
+DB_TYPE = "MYSQL"
+DESTINATION_GCS_FOLDER = "gs://{BUCKET}>"
+
+# Service definition
+# Docs: https://cloud.google.com/dataproc-metastore/docs/reference/rest/v1/projects.locations.services#Service
+# [START how_to_cloud_dataproc_metastore_create_service]
+SERVICE = {
+ "name": "test-service",
+}
+# [END how_to_cloud_dataproc_metastore_create_service]
+
+# Update service
+# [START how_to_cloud_dataproc_metastore_update_service]
+SERVICE_TO_UPDATE = {
+ "labels": {
+ "mylocalmachine": "mylocalmachine",
+ "systemtest": "systemtest",
+ }
+}
+UPDATE_MASK = {"paths": ["labels"]}
+# [END how_to_cloud_dataproc_metastore_update_service]
+
+# Backup definition
+# [START how_to_cloud_dataproc_metastore_create_backup]
+BACKUP = {
+ "name": "test-backup",
+}
+# [END how_to_cloud_dataproc_metastore_create_backup]
+
+# Metadata import definition
+# [START how_to_cloud_dataproc_metastore_create_metadata_import]
+METADATA_IMPORT = {
+ "name": "test-metadata-import",
+ "database_dump": {
+ "gcs_uri": GCS_URI,
+ "database_type": DB_TYPE,
+ },
+}
+# [END how_to_cloud_dataproc_metastore_create_metadata_import]
+
+
+with models.DAG("example_gcp_dataproc_metastore", start_date=days_ago(1), schedule_interval="@once") as dag:
+ # [START how_to_cloud_dataproc_metastore_create_service_operator]
+ create_service = DataprocMetastoreCreateServiceOperator(
+ task_id="create_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service=SERVICE,
+ service_id=SERVICE_ID,
+ timeout=TIMEOUT,
+ )
+ # [END how_to_cloud_dataproc_metastore_create_service_operator]
+
+ # [START how_to_cloud_dataproc_metastore_get_service_operator]
+ get_service_details = DataprocMetastoreGetServiceOperator(
+ task_id="get_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service_id=SERVICE_ID,
Review comment:
It looks like `project_id`, `region`, and `service_id` are passed to all of the operators with the same value. This would be a _great_ opportunity for adding these to `default_args` and removing them from each of the individual operators.
You may want to update the operator docs to make a note that these args have been passed used in `default_args` of the DAG so it's transparent why some args are obviously missing from the operator instantiations.
##########
File path: airflow/providers/google/cloud/example_dags/example_dataproc_metastore.py
##########
@@ -0,0 +1,202 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example Airflow DAG that show how to use various Dataproc Metastore
+operators to manage a service.
+"""
+
+import os
+
+from airflow import models
+from airflow.providers.google.cloud.operators.dataproc_metastore import (
+ DataprocMetastoreCreateBackupOperator,
+ DataprocMetastoreCreateMetadataImportOperator,
+ DataprocMetastoreCreateServiceOperator,
+ DataprocMetastoreDeleteBackupOperator,
+ DataprocMetastoreDeleteServiceOperator,
+ DataprocMetastoreExportMetadataOperator,
+ DataprocMetastoreGetServiceOperator,
+ DataprocMetastoreListBackupsOperator,
+ DataprocMetastoreRestoreServiceOperator,
+ DataprocMetastoreUpdateServiceOperator,
+)
+from airflow.utils.dates import days_ago
+
+PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "<PROJECT_ID>")
+SERVICE_ID = os.environ.get("GCP_DATAPROC_METASTORE_SERVICE_ID", "dataproc-metastore-system-tests-service-1")
+BACKUP_ID = os.environ.get("GCP_DATAPROC_METASTORE_BACKUP_ID", "dataproc-metastore-system-tests-backup-1")
+REGION = os.environ.get("GCP_REGION", "<REGION>")
+BUCKET = os.environ.get("GCP_DATAPROC_METASTORE_BUCKET", "dataproc-metastore-system-tests")
+METADATA_IMPORT_FILE = os.environ.get("GCS_METADATA_IMPORT_FILE", None)
+METADATA_IMPORT_ID = "dataproc-metastore-system-tests-metadata-import-1"
+TIMEOUT = 1200
+GCS_URI = f"gs://{BUCKET}/{METADATA_IMPORT_FILE}"
+DB_TYPE = "MYSQL"
+DESTINATION_GCS_FOLDER = "gs://{BUCKET}>"
+
+# Service definition
+# Docs: https://cloud.google.com/dataproc-metastore/docs/reference/rest/v1/projects.locations.services#Service
+# [START how_to_cloud_dataproc_metastore_create_service]
+SERVICE = {
+ "name": "test-service",
+}
+# [END how_to_cloud_dataproc_metastore_create_service]
+
+# Update service
+# [START how_to_cloud_dataproc_metastore_update_service]
+SERVICE_TO_UPDATE = {
+ "labels": {
+ "mylocalmachine": "mylocalmachine",
+ "systemtest": "systemtest",
+ }
+}
+UPDATE_MASK = {"paths": ["labels"]}
+# [END how_to_cloud_dataproc_metastore_update_service]
+
+# Backup definition
+# [START how_to_cloud_dataproc_metastore_create_backup]
+BACKUP = {
+ "name": "test-backup",
+}
+# [END how_to_cloud_dataproc_metastore_create_backup]
+
+# Metadata import definition
+# [START how_to_cloud_dataproc_metastore_create_metadata_import]
+METADATA_IMPORT = {
+ "name": "test-metadata-import",
+ "database_dump": {
+ "gcs_uri": GCS_URI,
+ "database_type": DB_TYPE,
+ },
+}
+# [END how_to_cloud_dataproc_metastore_create_metadata_import]
+
+
+with models.DAG("example_gcp_dataproc_metastore", start_date=days_ago(1), schedule_interval="@once") as dag:
+ # [START how_to_cloud_dataproc_metastore_create_service_operator]
+ create_service = DataprocMetastoreCreateServiceOperator(
+ task_id="create_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service=SERVICE,
+ service_id=SERVICE_ID,
+ timeout=TIMEOUT,
+ )
+ # [END how_to_cloud_dataproc_metastore_create_service_operator]
+
+ # [START how_to_cloud_dataproc_metastore_get_service_operator]
+ get_service_details = DataprocMetastoreGetServiceOperator(
+ task_id="get_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service_id=SERVICE_ID,
Review comment:
@mik-laj WDYT about this be a case where the documentation could have explicit code blocks rather than reference the example DAG? Seems like folks would look at the example DAG in its entirety, as a demonstration of a use case, while the documentation serves as the individual operator information.
I'm not saying we should never reference the example DAG in the operator docs but there might be cases where the deference to having a solid, well-written example DAG (especially ones showcasing a nice feature that simplifies authoring) makes sense.
##########
File path: airflow/providers/google/cloud/example_dags/example_dataproc_metastore.py
##########
@@ -0,0 +1,202 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example Airflow DAG that show how to use various Dataproc Metastore
+operators to manage a service.
+"""
+
+import os
+
+from airflow import models
+from airflow.providers.google.cloud.operators.dataproc_metastore import (
+ DataprocMetastoreCreateBackupOperator,
+ DataprocMetastoreCreateMetadataImportOperator,
+ DataprocMetastoreCreateServiceOperator,
+ DataprocMetastoreDeleteBackupOperator,
+ DataprocMetastoreDeleteServiceOperator,
+ DataprocMetastoreExportMetadataOperator,
+ DataprocMetastoreGetServiceOperator,
+ DataprocMetastoreListBackupsOperator,
+ DataprocMetastoreRestoreServiceOperator,
+ DataprocMetastoreUpdateServiceOperator,
+)
+from airflow.utils.dates import days_ago
+
+PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "<PROJECT_ID>")
+SERVICE_ID = os.environ.get("GCP_DATAPROC_METASTORE_SERVICE_ID", "dataproc-metastore-system-tests-service-1")
+BACKUP_ID = os.environ.get("GCP_DATAPROC_METASTORE_BACKUP_ID", "dataproc-metastore-system-tests-backup-1")
+REGION = os.environ.get("GCP_REGION", "<REGION>")
+BUCKET = os.environ.get("GCP_DATAPROC_METASTORE_BUCKET", "dataproc-metastore-system-tests")
+METADATA_IMPORT_FILE = os.environ.get("GCS_METADATA_IMPORT_FILE", None)
+METADATA_IMPORT_ID = "dataproc-metastore-system-tests-metadata-import-1"
+TIMEOUT = 1200
+GCS_URI = f"gs://{BUCKET}/{METADATA_IMPORT_FILE}"
+DB_TYPE = "MYSQL"
+DESTINATION_GCS_FOLDER = "gs://{BUCKET}>"
+
+# Service definition
+# Docs: https://cloud.google.com/dataproc-metastore/docs/reference/rest/v1/projects.locations.services#Service
+# [START how_to_cloud_dataproc_metastore_create_service]
+SERVICE = {
+ "name": "test-service",
+}
+# [END how_to_cloud_dataproc_metastore_create_service]
+
+# Update service
+# [START how_to_cloud_dataproc_metastore_update_service]
+SERVICE_TO_UPDATE = {
+ "labels": {
+ "mylocalmachine": "mylocalmachine",
+ "systemtest": "systemtest",
+ }
+}
+UPDATE_MASK = {"paths": ["labels"]}
+# [END how_to_cloud_dataproc_metastore_update_service]
+
+# Backup definition
+# [START how_to_cloud_dataproc_metastore_create_backup]
+BACKUP = {
+ "name": "test-backup",
+}
+# [END how_to_cloud_dataproc_metastore_create_backup]
+
+# Metadata import definition
+# [START how_to_cloud_dataproc_metastore_create_metadata_import]
+METADATA_IMPORT = {
+ "name": "test-metadata-import",
+ "database_dump": {
+ "gcs_uri": GCS_URI,
+ "database_type": DB_TYPE,
+ },
+}
+# [END how_to_cloud_dataproc_metastore_create_metadata_import]
+
+
+with models.DAG("example_gcp_dataproc_metastore", start_date=days_ago(1), schedule_interval="@once") as dag:
+ # [START how_to_cloud_dataproc_metastore_create_service_operator]
+ create_service = DataprocMetastoreCreateServiceOperator(
+ task_id="create_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service=SERVICE,
+ service_id=SERVICE_ID,
+ timeout=TIMEOUT,
+ )
+ # [END how_to_cloud_dataproc_metastore_create_service_operator]
+
+ # [START how_to_cloud_dataproc_metastore_get_service_operator]
+ get_service_details = DataprocMetastoreGetServiceOperator(
+ task_id="get_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service_id=SERVICE_ID,
Review comment:
@mik-laj WDYT about this being a case where the documentation could have explicit code blocks rather than reference the example DAG? Seems like folks would look at the example DAG in its entirety, as a demonstration of a use case, while the documentation serves as the individual operator information.
I'm not saying we should never reference the example DAG in the operator docs but there might be cases where the deference to having a solid, well-written example DAG (especially ones showcasing a nice feature that simplifies authoring) makes sense.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] lwyszomi commented on a change in pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
lwyszomi commented on a change in pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#discussion_r731683771
##########
File path: airflow/providers/google/cloud/hooks/dataproc_metastore.py
##########
@@ -0,0 +1,668 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+"""This module contains a Google Cloud Dataproc Metastore hook."""
+
+from typing import Dict, Optional, Sequence, Tuple, Union
+
+from google.api_core.operation import Operation
+from google.api_core.retry import Retry
+from google.cloud.metastore_v1 import DataprocMetastoreClient
+from google.cloud.metastore_v1.types import Backup, MetadataImport, Service
+from google.cloud.metastore_v1.types.metastore import DatabaseDumpSpec, Restore
+from google.protobuf.field_mask_pb2 import FieldMask
+
+from airflow.exceptions import AirflowException
+from airflow.providers.google.common.hooks.base_google import GoogleBaseHook
+
+
+class DataprocMetastoreHook(GoogleBaseHook):
+ """Hook for Google Cloud Dataproc Metastore APIs."""
+
+ def get_dataproc_metastore_client(self, region: Optional[str] = None) -> DataprocMetastoreClient:
+ """Returns DataprocMetastoreClient."""
+ client_options = None
+ if region and region != 'global':
+ client_options = {'api_endpoint': f'{region}-metastore.googleapis.com:443'}
+
+ return DataprocMetastoreClient(
+ credentials=self._get_credentials(), client_info=self.client_info, client_options=client_options
+ )
+
+ def wait_for_operation(self, timeout: float, operation: Operation):
+ """Waits for long-lasting operation to complete."""
+ error = operation.exception(timeout=timeout)
Review comment:
@wojsamjan This is not clear what happens here (from logic perspective 😄 ) maybe to make code more readable we can use this sample:
```
try:
return operation.result(timeout=timeout)
except Exception:
error = operaion.exception()
raise AirflowException(error)
```
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mik-laj commented on a change in pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#discussion_r728417941
##########
File path: airflow/providers/google/cloud/example_dags/example_dataproc_metastore.py
##########
@@ -0,0 +1,202 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example Airflow DAG that show how to use various Dataproc Metastore
+operators to manage a service.
+"""
+
+import os
+
+from airflow import models
+from airflow.providers.google.cloud.operators.dataproc_metastore import (
+ DataprocMetastoreCreateBackupOperator,
+ DataprocMetastoreCreateMetadataImportOperator,
+ DataprocMetastoreCreateServiceOperator,
+ DataprocMetastoreDeleteBackupOperator,
+ DataprocMetastoreDeleteServiceOperator,
+ DataprocMetastoreExportMetadataOperator,
+ DataprocMetastoreGetServiceOperator,
+ DataprocMetastoreListBackupsOperator,
+ DataprocMetastoreRestoreServiceOperator,
+ DataprocMetastoreUpdateServiceOperator,
+)
+from airflow.utils.dates import days_ago
+
+PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "<PROJECT_ID>")
+SERVICE_ID = os.environ.get("GCP_DATAPROC_METASTORE_SERVICE_ID", "dataproc-metastore-system-tests-service-1")
+BACKUP_ID = os.environ.get("GCP_DATAPROC_METASTORE_BACKUP_ID", "dataproc-metastore-system-tests-backup-1")
+REGION = os.environ.get("GCP_REGION", "<REGION>")
+BUCKET = os.environ.get("GCP_DATAPROC_METASTORE_BUCKET", "dataproc-metastore-system-tests")
+METADATA_IMPORT_FILE = os.environ.get("GCS_METADATA_IMPORT_FILE", None)
+METADATA_IMPORT_ID = "dataproc-metastore-system-tests-metadata-import-1"
+TIMEOUT = 1200
+GCS_URI = f"gs://{BUCKET}/{METADATA_IMPORT_FILE}"
+DB_TYPE = "MYSQL"
+DESTINATION_GCS_FOLDER = "gs://{BUCKET}>"
+
+# Service definition
+# Docs: https://cloud.google.com/dataproc-metastore/docs/reference/rest/v1/projects.locations.services#Service
+# [START how_to_cloud_dataproc_metastore_create_service]
+SERVICE = {
+ "name": "test-service",
+}
+# [END how_to_cloud_dataproc_metastore_create_service]
+
+# Update service
+# [START how_to_cloud_dataproc_metastore_update_service]
+SERVICE_TO_UPDATE = {
+ "labels": {
+ "mylocalmachine": "mylocalmachine",
+ "systemtest": "systemtest",
+ }
+}
+UPDATE_MASK = {"paths": ["labels"]}
+# [END how_to_cloud_dataproc_metastore_update_service]
+
+# Backup definition
+# [START how_to_cloud_dataproc_metastore_create_backup]
+BACKUP = {
+ "name": "test-backup",
+}
+# [END how_to_cloud_dataproc_metastore_create_backup]
+
+# Metadata import definition
+# [START how_to_cloud_dataproc_metastore_create_metadata_import]
+METADATA_IMPORT = {
+ "name": "test-metadata-import",
+ "database_dump": {
+ "gcs_uri": GCS_URI,
+ "database_type": DB_TYPE,
+ },
+}
+# [END how_to_cloud_dataproc_metastore_create_metadata_import]
+
+
+with models.DAG("example_gcp_dataproc_metastore", start_date=days_ago(1), schedule_interval="@once") as dag:
+ # [START how_to_cloud_dataproc_metastore_create_service_operator]
+ create_service = DataprocMetastoreCreateServiceOperator(
+ task_id="create_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service=SERVICE,
+ service_id=SERVICE_ID,
+ timeout=TIMEOUT,
+ )
+ # [END how_to_cloud_dataproc_metastore_create_service_operator]
+
+ # [START how_to_cloud_dataproc_metastore_get_service_operator]
+ get_service_details = DataprocMetastoreGetServiceOperator(
+ task_id="get_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service_id=SERVICE_ID,
Review comment:
This will work as long as the job definition is in a separate Python file. To add examples to documentation, we use `example-include` directive, To allow us to keep docs and example dags in sync
https://github.com/apache/airflow/blob/af4a5e006e4f5c9f203afeac039b22c6adee317f/docs/apache-airflow-providers-google/operators/cloud/dataproc.rst#L54-L58
Alternatively, we can also use the `code-block` directive, but this means that the example will not be tested and executed on CI. If it is not running on a CI, the examples very quickly become out-of-date and are more of a burden to maintain.
https://github.com/apache/airflow/blob/060345c/docs/apache-airflow/tutorial.rst#L205-L231
For example:
In [this PR](https://github.com/apache/airflow/pull/18036/files), I deprecated default value of `pod_name` parameter in `EKSCreateClusterOperator`. This meant that if someone loads this file `DeprecationWarning` will be displayed. But we also [detect such problems](https://github.com/apache/airflow/pull/17900) automatically on CI, and we won't allow a merge PR that doesn't update the examples. This makes these examples much easier to maintain. We don't have to worry if all the examples have been updated to show the best practices because we detect all such problems automatically and if something still needs to be changed, we will inform everyone on the CI. If we used the `code-block` directive, we would have to find a way to detect similar problems automatically as well.
This is one example, but we have other automated checks that allow us to take care of the high quality of documentation and examples
> Just thought it was an opportunity for some legitimate use around default_args and document it.
I agree that this is a good use for `default_args`, but given the problems with using these examples in the documentation, we don't necessarily have to do.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] josh-fell commented on a change in pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
josh-fell commented on a change in pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#discussion_r728128151
##########
File path: airflow/providers/google/cloud/example_dags/example_dataproc_metastore.py
##########
@@ -0,0 +1,202 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example Airflow DAG that show how to use various Dataproc Metastore
+operators to manage a service.
+"""
+
+import os
+
+from airflow import models
+from airflow.providers.google.cloud.operators.dataproc_metastore import (
+ DataprocMetastoreCreateBackupOperator,
+ DataprocMetastoreCreateMetadataImportOperator,
+ DataprocMetastoreCreateServiceOperator,
+ DataprocMetastoreDeleteBackupOperator,
+ DataprocMetastoreDeleteServiceOperator,
+ DataprocMetastoreExportMetadataOperator,
+ DataprocMetastoreGetServiceOperator,
+ DataprocMetastoreListBackupsOperator,
+ DataprocMetastoreRestoreServiceOperator,
+ DataprocMetastoreUpdateServiceOperator,
+)
+from airflow.utils.dates import days_ago
+
+PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "<PROJECT_ID>")
+SERVICE_ID = os.environ.get("GCP_DATAPROC_METASTORE_SERVICE_ID", "dataproc-metastore-system-tests-service-1")
+BACKUP_ID = os.environ.get("GCP_DATAPROC_METASTORE_BACKUP_ID", "dataproc-metastore-system-tests-backup-1")
+REGION = os.environ.get("GCP_REGION", "<REGION>")
+BUCKET = os.environ.get("GCP_DATAPROC_METASTORE_BUCKET", "dataproc-metastore-system-tests")
+METADATA_IMPORT_FILE = os.environ.get("GCS_METADATA_IMPORT_FILE", None)
+METADATA_IMPORT_ID = "dataproc-metastore-system-tests-metadata-import-1"
+TIMEOUT = 1200
+GCS_URI = f"gs://{BUCKET}/{METADATA_IMPORT_FILE}"
+DB_TYPE = "MYSQL"
+DESTINATION_GCS_FOLDER = "gs://{BUCKET}>"
+
+# Service definition
+# Docs: https://cloud.google.com/dataproc-metastore/docs/reference/rest/v1/projects.locations.services#Service
+# [START how_to_cloud_dataproc_metastore_create_service]
+SERVICE = {
+ "name": "test-service",
+}
+# [END how_to_cloud_dataproc_metastore_create_service]
+
+# Update service
+# [START how_to_cloud_dataproc_metastore_update_service]
+SERVICE_TO_UPDATE = {
+ "labels": {
+ "mylocalmachine": "mylocalmachine",
+ "systemtest": "systemtest",
+ }
+}
+UPDATE_MASK = {"paths": ["labels"]}
+# [END how_to_cloud_dataproc_metastore_update_service]
+
+# Backup definition
+# [START how_to_cloud_dataproc_metastore_create_backup]
+BACKUP = {
+ "name": "test-backup",
+}
+# [END how_to_cloud_dataproc_metastore_create_backup]
+
+# Metadata import definition
+# [START how_to_cloud_dataproc_metastore_create_metadata_import]
+METADATA_IMPORT = {
+ "name": "test-metadata-import",
+ "database_dump": {
+ "gcs_uri": GCS_URI,
+ "database_type": DB_TYPE,
+ },
+}
+# [END how_to_cloud_dataproc_metastore_create_metadata_import]
+
+
+with models.DAG("example_gcp_dataproc_metastore", start_date=days_ago(1), schedule_interval="@once") as dag:
Review comment:
There is an ongoing effort to update the example DAGs in Airflow to transition away from `days_ago(n)` for `start_date` to a static value, improve `default_args` use, etc. when applicable. For new DAGs, let's follow suit for using a static `start_date` value as it is best practice.
##########
File path: airflow/providers/google/cloud/example_dags/example_dataproc_metastore.py
##########
@@ -0,0 +1,202 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example Airflow DAG that show how to use various Dataproc Metastore
+operators to manage a service.
+"""
+
+import os
+
+from airflow import models
+from airflow.providers.google.cloud.operators.dataproc_metastore import (
+ DataprocMetastoreCreateBackupOperator,
+ DataprocMetastoreCreateMetadataImportOperator,
+ DataprocMetastoreCreateServiceOperator,
+ DataprocMetastoreDeleteBackupOperator,
+ DataprocMetastoreDeleteServiceOperator,
+ DataprocMetastoreExportMetadataOperator,
+ DataprocMetastoreGetServiceOperator,
+ DataprocMetastoreListBackupsOperator,
+ DataprocMetastoreRestoreServiceOperator,
+ DataprocMetastoreUpdateServiceOperator,
+)
+from airflow.utils.dates import days_ago
+
+PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "<PROJECT_ID>")
+SERVICE_ID = os.environ.get("GCP_DATAPROC_METASTORE_SERVICE_ID", "dataproc-metastore-system-tests-service-1")
+BACKUP_ID = os.environ.get("GCP_DATAPROC_METASTORE_BACKUP_ID", "dataproc-metastore-system-tests-backup-1")
+REGION = os.environ.get("GCP_REGION", "<REGION>")
+BUCKET = os.environ.get("GCP_DATAPROC_METASTORE_BUCKET", "dataproc-metastore-system-tests")
+METADATA_IMPORT_FILE = os.environ.get("GCS_METADATA_IMPORT_FILE", None)
+METADATA_IMPORT_ID = "dataproc-metastore-system-tests-metadata-import-1"
+TIMEOUT = 1200
+GCS_URI = f"gs://{BUCKET}/{METADATA_IMPORT_FILE}"
+DB_TYPE = "MYSQL"
+DESTINATION_GCS_FOLDER = "gs://{BUCKET}>"
+
+# Service definition
+# Docs: https://cloud.google.com/dataproc-metastore/docs/reference/rest/v1/projects.locations.services#Service
+# [START how_to_cloud_dataproc_metastore_create_service]
+SERVICE = {
+ "name": "test-service",
+}
+# [END how_to_cloud_dataproc_metastore_create_service]
+
+# Update service
+# [START how_to_cloud_dataproc_metastore_update_service]
+SERVICE_TO_UPDATE = {
+ "labels": {
+ "mylocalmachine": "mylocalmachine",
+ "systemtest": "systemtest",
+ }
+}
+UPDATE_MASK = {"paths": ["labels"]}
+# [END how_to_cloud_dataproc_metastore_update_service]
+
+# Backup definition
+# [START how_to_cloud_dataproc_metastore_create_backup]
+BACKUP = {
+ "name": "test-backup",
+}
+# [END how_to_cloud_dataproc_metastore_create_backup]
+
+# Metadata import definition
+# [START how_to_cloud_dataproc_metastore_create_metadata_import]
+METADATA_IMPORT = {
+ "name": "test-metadata-import",
+ "database_dump": {
+ "gcs_uri": GCS_URI,
+ "database_type": DB_TYPE,
+ },
+}
+# [END how_to_cloud_dataproc_metastore_create_metadata_import]
+
+
+with models.DAG("example_gcp_dataproc_metastore", start_date=days_ago(1), schedule_interval="@once") as dag:
+ # [START how_to_cloud_dataproc_metastore_create_service_operator]
+ create_service = DataprocMetastoreCreateServiceOperator(
+ task_id="create_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service=SERVICE,
+ service_id=SERVICE_ID,
+ timeout=TIMEOUT,
+ )
+ # [END how_to_cloud_dataproc_metastore_create_service_operator]
+
+ # [START how_to_cloud_dataproc_metastore_get_service_operator]
+ get_service_details = DataprocMetastoreGetServiceOperator(
+ task_id="get_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service_id=SERVICE_ID,
+ )
+ # [END how_to_cloud_dataproc_metastore_get_service_operator]
+
+ # [START how_to_cloud_dataproc_metastore_update_service_operator]
+ update_service = DataprocMetastoreUpdateServiceOperator(
+ task_id="update_service",
+ project_id=PROJECT_ID,
+ service_id=SERVICE_ID,
+ region=REGION,
+ service=SERVICE_TO_UPDATE,
+ update_mask=UPDATE_MASK,
+ )
+ # [END how_to_cloud_dataproc_metastore_update_service_operator]
+
+ # [START how_to_cloud_dataproc_metastore_create_metadata_import_operator]
+ import_metadata = DataprocMetastoreCreateMetadataImportOperator(
+ task_id="create_metadata_import",
+ project_id=PROJECT_ID,
+ region=REGION,
+ service_id=SERVICE_ID,
+ metadata_import=METADATA_IMPORT,
+ metadata_import_id=METADATA_IMPORT_ID,
+ )
+ # [END how_to_cloud_dataproc_metastore_create_metadata_import_operator]
+
+ # [START how_to_cloud_dataproc_metastore_export_metadata_operator]
+ export_metadata = DataprocMetastoreExportMetadataOperator(
+ task_id="export_metadata",
+ destination_gcs_folder=DESTINATION_GCS_FOLDER,
+ project_id=PROJECT_ID,
+ region=REGION,
+ service_id=SERVICE_ID,
+ timeout=TIMEOUT,
+ )
+ # [END how_to_cloud_dataproc_metastore_export_metadata_operator]
+
+ # [START how_to_cloud_dataproc_metastore_create_backup_operator]
+ backup_service = DataprocMetastoreCreateBackupOperator(
+ task_id="create_backup",
+ project_id=PROJECT_ID,
+ region=REGION,
+ service_id=SERVICE_ID,
+ backup=BACKUP,
+ backup_id=BACKUP_ID,
+ )
+ # [END how_to_cloud_dataproc_metastore_create_backup_operator]
+
+ # [START how_to_cloud_dataproc_metastore_list_backups_operator]
+ list_backups = DataprocMetastoreListBackupsOperator(
+ task_id="list_backups",
+ project_id=PROJECT_ID,
+ region=REGION,
+ service_id=SERVICE_ID,
+ )
+ # [END how_to_cloud_dataproc_metastore_list_backups_operator]
+
+ # [START how_to_cloud_dataproc_metastore_delete_backup_operator]
+ delete_backup = DataprocMetastoreDeleteBackupOperator(
+ task_id="delete_backup",
+ project_id=PROJECT_ID,
+ region=REGION,
+ service_id=SERVICE_ID,
+ backup_id=BACKUP_ID,
+ )
+ # [END how_to_cloud_dataproc_metastore_delete_backup_operator]
+
+ # [START how_to_cloud_dataproc_metastore_restore_service_operator]
+ restore_service = DataprocMetastoreRestoreServiceOperator(
+ task_id="restore_metastore",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service_id=SERVICE_ID,
+ backup_id=BACKUP_ID,
+ backup_region=REGION,
+ backup_project_id=PROJECT_ID,
+ backup_service_id=SERVICE_ID,
+ )
+ # [END how_to_cloud_dataproc_metastore_restore_service_operator]
+
+ # [START how_to_cloud_dataproc_metastore_delete_service_operator]
+ delete_service = DataprocMetastoreDeleteServiceOperator(
+ task_id="delete_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service_id=SERVICE_ID,
+ )
+ # [END how_to_cloud_dataproc_metastore_delete_service_operator]
+
+ create_service >> update_service >> get_service_details
+
+ get_service_details >> backup_service >> list_backups >> restore_service >> delete_backup
+
+ delete_backup >> export_metadata >> import_metadata
+
+ import_metadata >> delete_service
Review comment:
WDYT about using `airflow.models.baseoperator.chain()` here? It may be more readable to do something like
```python
chain(
create_service,
update_service,
get_service_details,
backup_service,
list_backups,
restore_service,
delete_backup,
export_metadata,
import_metadata,
delete_service,
)
```
##########
File path: airflow/providers/google/cloud/example_dags/example_dataproc_metastore.py
##########
@@ -0,0 +1,202 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example Airflow DAG that show how to use various Dataproc Metastore
+operators to manage a service.
+"""
+
+import os
+
+from airflow import models
+from airflow.providers.google.cloud.operators.dataproc_metastore import (
+ DataprocMetastoreCreateBackupOperator,
+ DataprocMetastoreCreateMetadataImportOperator,
+ DataprocMetastoreCreateServiceOperator,
+ DataprocMetastoreDeleteBackupOperator,
+ DataprocMetastoreDeleteServiceOperator,
+ DataprocMetastoreExportMetadataOperator,
+ DataprocMetastoreGetServiceOperator,
+ DataprocMetastoreListBackupsOperator,
+ DataprocMetastoreRestoreServiceOperator,
+ DataprocMetastoreUpdateServiceOperator,
+)
+from airflow.utils.dates import days_ago
+
+PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "<PROJECT_ID>")
+SERVICE_ID = os.environ.get("GCP_DATAPROC_METASTORE_SERVICE_ID", "dataproc-metastore-system-tests-service-1")
+BACKUP_ID = os.environ.get("GCP_DATAPROC_METASTORE_BACKUP_ID", "dataproc-metastore-system-tests-backup-1")
+REGION = os.environ.get("GCP_REGION", "<REGION>")
+BUCKET = os.environ.get("GCP_DATAPROC_METASTORE_BUCKET", "dataproc-metastore-system-tests")
+METADATA_IMPORT_FILE = os.environ.get("GCS_METADATA_IMPORT_FILE", None)
+METADATA_IMPORT_ID = "dataproc-metastore-system-tests-metadata-import-1"
+TIMEOUT = 1200
+GCS_URI = f"gs://{BUCKET}/{METADATA_IMPORT_FILE}"
+DB_TYPE = "MYSQL"
+DESTINATION_GCS_FOLDER = "gs://{BUCKET}>"
+
+# Service definition
+# Docs: https://cloud.google.com/dataproc-metastore/docs/reference/rest/v1/projects.locations.services#Service
+# [START how_to_cloud_dataproc_metastore_create_service]
+SERVICE = {
+ "name": "test-service",
+}
+# [END how_to_cloud_dataproc_metastore_create_service]
+
+# Update service
+# [START how_to_cloud_dataproc_metastore_update_service]
+SERVICE_TO_UPDATE = {
+ "labels": {
+ "mylocalmachine": "mylocalmachine",
+ "systemtest": "systemtest",
+ }
+}
+UPDATE_MASK = {"paths": ["labels"]}
+# [END how_to_cloud_dataproc_metastore_update_service]
+
+# Backup definition
+# [START how_to_cloud_dataproc_metastore_create_backup]
+BACKUP = {
+ "name": "test-backup",
+}
+# [END how_to_cloud_dataproc_metastore_create_backup]
+
+# Metadata import definition
+# [START how_to_cloud_dataproc_metastore_create_metadata_import]
+METADATA_IMPORT = {
+ "name": "test-metadata-import",
+ "database_dump": {
+ "gcs_uri": GCS_URI,
+ "database_type": DB_TYPE,
+ },
+}
+# [END how_to_cloud_dataproc_metastore_create_metadata_import]
+
+
+with models.DAG("example_gcp_dataproc_metastore", start_date=days_ago(1), schedule_interval="@once") as dag:
+ # [START how_to_cloud_dataproc_metastore_create_service_operator]
+ create_service = DataprocMetastoreCreateServiceOperator(
+ task_id="create_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service=SERVICE,
+ service_id=SERVICE_ID,
+ timeout=TIMEOUT,
+ )
+ # [END how_to_cloud_dataproc_metastore_create_service_operator]
+
+ # [START how_to_cloud_dataproc_metastore_get_service_operator]
+ get_service_details = DataprocMetastoreGetServiceOperator(
+ task_id="get_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service_id=SERVICE_ID,
Review comment:
It looks like `project_id`, `region`, and `service_id` are passed to all of the operators with the same value. This would be a _great_ opportunity for adding these to `default_args` and removing them from each of the individual operators.
You may want to update the operator docs to make a note that these args have been passed used in `default_args` of the DAG so it's transparent why some args are obviously missing from the operator instantiations.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] josh-fell commented on a change in pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
josh-fell commented on a change in pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#discussion_r728288602
##########
File path: airflow/providers/google/cloud/example_dags/example_dataproc_metastore.py
##########
@@ -0,0 +1,202 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example Airflow DAG that show how to use various Dataproc Metastore
+operators to manage a service.
+"""
+
+import os
+
+from airflow import models
+from airflow.providers.google.cloud.operators.dataproc_metastore import (
+ DataprocMetastoreCreateBackupOperator,
+ DataprocMetastoreCreateMetadataImportOperator,
+ DataprocMetastoreCreateServiceOperator,
+ DataprocMetastoreDeleteBackupOperator,
+ DataprocMetastoreDeleteServiceOperator,
+ DataprocMetastoreExportMetadataOperator,
+ DataprocMetastoreGetServiceOperator,
+ DataprocMetastoreListBackupsOperator,
+ DataprocMetastoreRestoreServiceOperator,
+ DataprocMetastoreUpdateServiceOperator,
+)
+from airflow.utils.dates import days_ago
+
+PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "<PROJECT_ID>")
+SERVICE_ID = os.environ.get("GCP_DATAPROC_METASTORE_SERVICE_ID", "dataproc-metastore-system-tests-service-1")
+BACKUP_ID = os.environ.get("GCP_DATAPROC_METASTORE_BACKUP_ID", "dataproc-metastore-system-tests-backup-1")
+REGION = os.environ.get("GCP_REGION", "<REGION>")
+BUCKET = os.environ.get("GCP_DATAPROC_METASTORE_BUCKET", "dataproc-metastore-system-tests")
+METADATA_IMPORT_FILE = os.environ.get("GCS_METADATA_IMPORT_FILE", None)
+METADATA_IMPORT_ID = "dataproc-metastore-system-tests-metadata-import-1"
+TIMEOUT = 1200
+GCS_URI = f"gs://{BUCKET}/{METADATA_IMPORT_FILE}"
+DB_TYPE = "MYSQL"
+DESTINATION_GCS_FOLDER = "gs://{BUCKET}>"
+
+# Service definition
+# Docs: https://cloud.google.com/dataproc-metastore/docs/reference/rest/v1/projects.locations.services#Service
+# [START how_to_cloud_dataproc_metastore_create_service]
+SERVICE = {
+ "name": "test-service",
+}
+# [END how_to_cloud_dataproc_metastore_create_service]
+
+# Update service
+# [START how_to_cloud_dataproc_metastore_update_service]
+SERVICE_TO_UPDATE = {
+ "labels": {
+ "mylocalmachine": "mylocalmachine",
+ "systemtest": "systemtest",
+ }
+}
+UPDATE_MASK = {"paths": ["labels"]}
+# [END how_to_cloud_dataproc_metastore_update_service]
+
+# Backup definition
+# [START how_to_cloud_dataproc_metastore_create_backup]
+BACKUP = {
+ "name": "test-backup",
+}
+# [END how_to_cloud_dataproc_metastore_create_backup]
+
+# Metadata import definition
+# [START how_to_cloud_dataproc_metastore_create_metadata_import]
+METADATA_IMPORT = {
+ "name": "test-metadata-import",
+ "database_dump": {
+ "gcs_uri": GCS_URI,
+ "database_type": DB_TYPE,
+ },
+}
+# [END how_to_cloud_dataproc_metastore_create_metadata_import]
+
+
+with models.DAG("example_gcp_dataproc_metastore", start_date=days_ago(1), schedule_interval="@once") as dag:
+ # [START how_to_cloud_dataproc_metastore_create_service_operator]
+ create_service = DataprocMetastoreCreateServiceOperator(
+ task_id="create_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service=SERVICE,
+ service_id=SERVICE_ID,
+ timeout=TIMEOUT,
+ )
+ # [END how_to_cloud_dataproc_metastore_create_service_operator]
+
+ # [START how_to_cloud_dataproc_metastore_get_service_operator]
+ get_service_details = DataprocMetastoreGetServiceOperator(
+ task_id="get_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service_id=SERVICE_ID,
Review comment:
@mik-laj WDYT about this be a case where the documentation could have explicit code blocks rather than reference the example DAG? Seems like folks would look at the example DAG in its entirety, as a demonstration of a use case, while the documentation serves as the individual operator information.
I'm not saying we should never reference the example DAG in the operator docs but there might be cases where the deference to having a solid, well-written example DAG (especially ones showcasing a nice feature that simplifies authoring) makes sense.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] wojsamjan commented on a change in pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
wojsamjan commented on a change in pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#discussion_r728882220
##########
File path: airflow/providers/google/cloud/hooks/dataproc_metastore.py
##########
@@ -0,0 +1,673 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+"""This module contains a Google Cloud Dataproc Metastore hook."""
+
+from time import sleep
+from typing import Dict, Optional, Sequence, Tuple, Union
+
+from google.api_core.operation import Operation
+from google.api_core.retry import Retry, exponential_sleep_generator
+from google.cloud.metastore_v1 import DataprocMetastoreClient
+from google.cloud.metastore_v1.types import Backup, MetadataImport, Service
+from google.cloud.metastore_v1.types.metastore import DatabaseDumpSpec, Restore
+from google.protobuf.field_mask_pb2 import FieldMask
+
+from airflow.exceptions import AirflowException
+from airflow.providers.google.common.hooks.base_google import GoogleBaseHook
+
+
+class DataprocMetastoreHook(GoogleBaseHook):
+ """Hook for Google Cloud Dataproc Metastore APIs."""
+
+ def get_dataproc_metastore_client(self, region: Optional[str] = None) -> DataprocMetastoreClient:
+ """Returns DataprocMetastoreClient."""
+ client_options = None
+ if region and region != 'global':
+ client_options = {'api_endpoint': f'{region}-metastore.googleapis.com:443'}
+
+ return DataprocMetastoreClient(
+ credentials=self._get_credentials(), client_info=self.client_info, client_options=client_options
+ )
+
+ def wait_for_operation(self, operation: Operation):
+ """Waits for long-lasting operation to complete."""
+ for time_to_wait in exponential_sleep_generator(initial=10, maximum=120):
+ sleep(time_to_wait)
+ if operation.done():
+ break
Review comment:
As I understand after the change I will have to pass timeout every time to the Hook´s wait_for_operation so both https://github.com/googleapis/python-api-core/blob/main/google/api_core/future/polling.py#L116 & https://github.com/googleapis/python-api-core/blob/main/google/api_core/future/polling.py#L141 could call https://github.com/googleapis/python-api-core/blob/main/google/api_core/future/polling.py#L106 with timeout passed for Retry´s deadline, right?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] josh-fell commented on a change in pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
josh-fell commented on a change in pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#discussion_r728288602
##########
File path: airflow/providers/google/cloud/example_dags/example_dataproc_metastore.py
##########
@@ -0,0 +1,202 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example Airflow DAG that show how to use various Dataproc Metastore
+operators to manage a service.
+"""
+
+import os
+
+from airflow import models
+from airflow.providers.google.cloud.operators.dataproc_metastore import (
+ DataprocMetastoreCreateBackupOperator,
+ DataprocMetastoreCreateMetadataImportOperator,
+ DataprocMetastoreCreateServiceOperator,
+ DataprocMetastoreDeleteBackupOperator,
+ DataprocMetastoreDeleteServiceOperator,
+ DataprocMetastoreExportMetadataOperator,
+ DataprocMetastoreGetServiceOperator,
+ DataprocMetastoreListBackupsOperator,
+ DataprocMetastoreRestoreServiceOperator,
+ DataprocMetastoreUpdateServiceOperator,
+)
+from airflow.utils.dates import days_ago
+
+PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "<PROJECT_ID>")
+SERVICE_ID = os.environ.get("GCP_DATAPROC_METASTORE_SERVICE_ID", "dataproc-metastore-system-tests-service-1")
+BACKUP_ID = os.environ.get("GCP_DATAPROC_METASTORE_BACKUP_ID", "dataproc-metastore-system-tests-backup-1")
+REGION = os.environ.get("GCP_REGION", "<REGION>")
+BUCKET = os.environ.get("GCP_DATAPROC_METASTORE_BUCKET", "dataproc-metastore-system-tests")
+METADATA_IMPORT_FILE = os.environ.get("GCS_METADATA_IMPORT_FILE", None)
+METADATA_IMPORT_ID = "dataproc-metastore-system-tests-metadata-import-1"
+TIMEOUT = 1200
+GCS_URI = f"gs://{BUCKET}/{METADATA_IMPORT_FILE}"
+DB_TYPE = "MYSQL"
+DESTINATION_GCS_FOLDER = "gs://{BUCKET}>"
+
+# Service definition
+# Docs: https://cloud.google.com/dataproc-metastore/docs/reference/rest/v1/projects.locations.services#Service
+# [START how_to_cloud_dataproc_metastore_create_service]
+SERVICE = {
+ "name": "test-service",
+}
+# [END how_to_cloud_dataproc_metastore_create_service]
+
+# Update service
+# [START how_to_cloud_dataproc_metastore_update_service]
+SERVICE_TO_UPDATE = {
+ "labels": {
+ "mylocalmachine": "mylocalmachine",
+ "systemtest": "systemtest",
+ }
+}
+UPDATE_MASK = {"paths": ["labels"]}
+# [END how_to_cloud_dataproc_metastore_update_service]
+
+# Backup definition
+# [START how_to_cloud_dataproc_metastore_create_backup]
+BACKUP = {
+ "name": "test-backup",
+}
+# [END how_to_cloud_dataproc_metastore_create_backup]
+
+# Metadata import definition
+# [START how_to_cloud_dataproc_metastore_create_metadata_import]
+METADATA_IMPORT = {
+ "name": "test-metadata-import",
+ "database_dump": {
+ "gcs_uri": GCS_URI,
+ "database_type": DB_TYPE,
+ },
+}
+# [END how_to_cloud_dataproc_metastore_create_metadata_import]
+
+
+with models.DAG("example_gcp_dataproc_metastore", start_date=days_ago(1), schedule_interval="@once") as dag:
+ # [START how_to_cloud_dataproc_metastore_create_service_operator]
+ create_service = DataprocMetastoreCreateServiceOperator(
+ task_id="create_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service=SERVICE,
+ service_id=SERVICE_ID,
+ timeout=TIMEOUT,
+ )
+ # [END how_to_cloud_dataproc_metastore_create_service_operator]
+
+ # [START how_to_cloud_dataproc_metastore_get_service_operator]
+ get_service_details = DataprocMetastoreGetServiceOperator(
+ task_id="get_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service_id=SERVICE_ID,
Review comment:
@mik-laj WDYT about this being a case where the documentation could have explicit code blocks rather than reference the example DAG? Seems like folks would look at the example DAG in its entirety, as a demonstration of a use case, while the documentation serves as the individual operator information.
I'm not saying we should never reference the example DAG in the operator docs but there might be cases where the deference to having a solid, well-written example DAG (especially ones showcasing a nice feature that simplifies authoring) makes sense.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mik-laj edited a comment on pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
mik-laj edited a comment on pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#issuecomment-948179409
These problems looks related to this contribution
```
==== Providers postgres: 13 failures ====
tests/providers/google/cloud/hooks/test_dataproc_metastore.py::TestDataprocMetastoreWithDefaultProjectIdHook::test_create_service: AssertionError: Expected call: create_service(metadata=None, request={'parent': 'projects/test-project-id/locations/test-region', 'service_id': 'test-service-id', 'service': {}, 'request_id': None}, retry=None, timeout=None)
Actual call: create_service(metadata=(), request={'parent': 'projects/test-project-id/locations/test-region', 'service_id': 'test-service-id', 'service': {'name': 'test-service'}, 'request_id': None}, retry=None, timeout=None)
tests/providers/google/cloud/hooks/test_dataproc_metastore.py::TestDataprocMetastoreWithDefaultProjectIdHook::test_update_service: google.auth.exceptions.DefaultCredentialsError: Could not automatically determine credentials. Please set GOOGLE_APPLICATION_CREDENTIALS or explicitly create credentials and re-run the application. For more information, please see https://cloud.google.com/docs/authentication/getting-started
tests/providers/google/cloud/hooks/test_dataproc_metastore.py::TestDataprocMetastoreWithoutDefaultProjectIdHook::test_create_service: AssertionError: Expected call: create_service(metadata=None, request={'parent': 'projects/test-project-id/locations/test-region', 'service_id': 'test-service-id', 'service': {'name': 'test-service'}, 'request_id': None}, retry=None, timeout=None)
Actual call: create_service(metadata=(), request={'parent': 'projects/test-project-id/locations/test-region', 'service_id': 'test-service-id', 'service': {'name': 'test-service'}, 'request_id': None}, retry=None, timeout=None)
tests/providers/google/cloud/hooks/test_dataproc_metastore.py::TestDataprocMetastoreWithoutDefaultProjectIdHook::test_update_service: google.auth.exceptions.DefaultCredentialsError: Could not automatically determine credentials. Please set GOOGLE_APPLICATION_CREDENTIALS or explicitly create credentials and re-run the application. For more information, please see https://cloud.google.com/docs/authentication/getting-started
tests/providers/google/cloud/operators/test_dataproc_metastore.py::TestDataprocMetastoreCreateBackupOperator::test_assert_valid_hook_call: TypeError: <MagicMock name='DataprocMetastoreHook().wait_for_operation()' id='140010347401960'> is not an instance of Backup
tests/providers/google/cloud/operators/test_dataproc_metastore.py::TestDataprocMetastoreCreateMetadataImportOperator::test_assert_valid_hook_call: TypeError: <MagicMock name='DataprocMetastoreHook().wait_for_operation()' id='140010430133976'> is not an instance of MetadataImport
tests/providers/google/cloud/operators/test_dataproc_metastore.py::TestDataprocMetastoreCreateServiceOperator::test_execute: TypeError: <MagicMock name='DataprocMetastoreHook().wait_for_operation()' id='140010332321160'> is not an instance of Service
tests/providers/google/cloud/operators/test_dataproc_metastore.py::TestDataprocMetastoreDeleteBackupOperator::test_assert_valid_hook_call: AssertionError: Expected call: DataprocMetastoreHook(gcp_conn_id='test-gcp-conn-id')
Actual call: DataprocMetastoreHook(gcp_conn_id='test-gcp-conn-id', impersonation_chain=None)
tests/providers/google/cloud/operators/test_dataproc_metastore.py::TestDataprocMetastoreExportMetadataOperator::test_assert_valid_hook_call: airflow.exceptions.AirflowException: Argument ['service_id'] is required
tests/providers/google/cloud/operators/test_dataproc_metastore.py::TestDataprocMetastoreGetServiceOperator::test_execute: TypeError: <MagicMock name='DataprocMetastoreHook().get_service()' id='140010395853488'> is not an instance of Service
tests/providers/google/cloud/operators/test_dataproc_metastore.py::TestDataprocMetastoreListBackupsOperator::test_assert_valid_hook_call: AssertionError: Expected call: DataprocMetastoreHook(gcp_conn_id='test-gcp-conn-id')
Actual call: DataprocMetastoreHook(gcp_conn_id='test-gcp-conn-id', impersonation_chain=None)
tests/providers/google/cloud/operators/test_dataproc_metastore.py::TestDataprocMetastoreRestoreServiceOperator::test_assert_valid_hook_call: Failed: Timeout >60.0s
tests/providers/google/cloud/operators/test_dataproc_metastore.py::TestDataprocMetastoreUpdateServiceOperator::test_assert_valid_hook_call: AssertionError: Expected call: DataprocMetastoreHook(gcp_conn_id='test-gcp-conn-id')
Actual call: DataprocMetastoreHook(gcp_conn_id='test-gcp-conn-id', impersonation_chain=None)
Error: Process completed with exit code 1.
```
https://github.com/apache/airflow/pull/18945/checks?check_run_id=3953505880
```
Run codespell to check for common misspellings in files..................................Failed
- hook id: codespell
- exit code: 65
airflow/providers/google/cloud/operators/dataproc_metastore.py:46: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/operators/dataproc_metastore.py:60: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/operators/dataproc_metastore.py:168: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/operators/dataproc_metastore.py:182: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/operators/dataproc_metastore.py:281: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/operators/dataproc_metastore.py:385: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/operators/dataproc_metastore.py:393: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/operators/dataproc_metastore.py:549: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/operators/dataproc_metastore.py:664: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/operators/dataproc_metastore.py:743: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/operators/dataproc_metastore.py:834: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/operators/dataproc_metastore.py:849: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/operators/dataproc_metastore.py:974: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/hooks/dataproc_metastore.py:76: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/hooks/dataproc_metastore.py:90: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/hooks/dataproc_metastore.py:142: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/hooks/dataproc_metastore.py:156: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/hooks/dataproc_metastore.py:214: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/hooks/dataproc_metastore.py:266: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/hooks/dataproc_metastore.py:274: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/hooks/dataproc_metastore.py:322: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/hooks/dataproc_metastore.py:378: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/hooks/dataproc_metastore.py:431: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/hooks/dataproc_metastore.py:480: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/hooks/dataproc_metastore.py:558: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/hooks/dataproc_metastore.py:573: alpha-numeric ==> alphanumeric
airflow/providers/google/cloud/hooks/dataproc_metastore.py:631: alpha-numeric ==> alphanumeric
```
https://github.com/apache/airflow/pull/18945/checks?check_run_id=3953505494
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] SamWheating commented on a change in pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
SamWheating commented on a change in pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#discussion_r728442051
##########
File path: airflow/providers/google/cloud/operators/dataproc_metastore.py
##########
@@ -0,0 +1,1046 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+"""This module contains Google Dataproc Metastore operators."""
+
+from time import sleep
+from typing import Dict, Optional, Sequence, Tuple, Union
+
+from google.api_core.retry import Retry, exponential_sleep_generator
+from google.cloud.metastore_v1 import MetadataExport, MetadataManagementActivity
+from google.cloud.metastore_v1.types import Backup, MetadataImport, Service
+from google.cloud.metastore_v1.types.metastore import DatabaseDumpSpec, Restore
+from google.protobuf.field_mask_pb2 import FieldMask
+from googleapiclient.errors import HttpError
+
+from airflow import AirflowException
+from airflow.models import BaseOperator
+from airflow.providers.google.cloud.hooks.dataproc_metastore import DataprocMetastoreHook
+
+
+class DataprocMetastoreCreateBackupOperator(BaseOperator):
+ """
+ Creates a new backup in a given project and location.
+
+ :param project_id: Required. The ID of the Google Cloud project that the service belongs to.
+ :type project_id: str
+ :param region: Required. The ID of the Google Cloud region that the service belongs to.
+ :type region: str
+ :param service_id: Required. The ID of the metastore service, which is used as the final component of
+ the metastore service's name. This value must be between 2 and 63 characters long inclusive, begin
+ with a letter, end with a letter or number, and consist of alpha-numeric ASCII characters or
+ hyphens.
+
+ This corresponds to the ``service_id`` field on the ``request`` instance; if ``request`` is
+ provided, this should not be set.
+ :type service_id: str
+ :param backup: Required. The backup to create. The ``name`` field is ignored. The ID of the created
+ backup must be provided in the request's ``backup_id`` field.
+
+ This corresponds to the ``backup`` field on the ``request`` instance; if ``request`` is provided, this
+ should not be set.
+ :type backup: google.cloud.metastore_v1.types.Backup
+ :param backup_id: Required. The ID of the backup, which is used as the final component of the backup's
+ name. This value must be between 1 and 64 characters long, begin with a letter, end with a letter or
+ number, and consist of alpha-numeric ASCII characters or hyphens.
+
+ This corresponds to the ``backup_id`` field on the ``request`` instance; if ``request`` is provided,
+ this should not be set.
+ :type backup_id: str
+ :param request_id: Optional. A unique id used to identify the request.
+ :type request_id: str
+ :param retry: Optional. Designation of what errors, if any, should be retried.
+ :type retry: google.api_core.retry.Retry
+ :param timeout: Optional. The timeout for this request.
+ :type timeout: float
+ :param metadata: Optional. Strings which should be sent along with the request as metadata.
+ :type metadata: Sequence[Tuple[str, str]]
+ :param gcp_conn_id: The connection ID to use connecting to Google Cloud.
+ :type gcp_conn_id: str
+ :param impersonation_chain: Optional service account to impersonate using short-term
+ credentials, or chained list of accounts required to get the access_token
+ of the last account in the list, which will be impersonated in the request.
+ If set as a string, the account must grant the originating account
+ the Service Account Token Creator IAM role.
+ If set as a sequence, the identities from the list must grant
+ Service Account Token Creator IAM role to the directly preceding identity, with first
+ account from the list granting this role to the originating account (templated).
+ :type impersonation_chain: Union[str, Sequence[str]]
+ """
+
+ template_fields = (
+ 'project_id',
+ 'backup',
+ 'impersonation_chain',
+ )
+ template_fields_renderers = {'backup': 'json'}
+
+ def __init__(
+ self,
+ *,
+ project_id: str,
+ region: str,
+ service_id: str,
+ backup: Union[Dict, Backup],
+ backup_id: str,
+ request_id: Optional[str] = None,
+ retry: Optional[Retry] = None,
+ timeout: Optional[float] = None,
+ metadata: Optional[Sequence[Tuple[str, str]]] = (),
+ gcp_conn_id: str = "google_cloud_default",
+ impersonation_chain: Optional[Union[str, Sequence[str]]] = None,
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.project_id = project_id
+ self.region = region
+ self.service_id = service_id
+ self.backup = backup
+ self.backup_id = backup_id
+ self.request_id = request_id
+ self.retry = retry
+ self.timeout = timeout
+ self.metadata = metadata
+ self.gcp_conn_id = gcp_conn_id
+ self.impersonation_chain = impersonation_chain
+
+ def execute(self, context: dict) -> dict:
+ hook = DataprocMetastoreHook(
+ gcp_conn_id=self.gcp_conn_id, impersonation_chain=self.impersonation_chain
+ )
+ self.log.info("Creating Dataproc Metastore backup: %s", self.backup_id)
+
+ try:
+ operation = hook.create_backup(
+ project_id=self.project_id,
+ region=self.region,
+ service_id=self.service_id,
+ backup=self.backup,
+ backup_id=self.backup_id,
+ request_id=self.request_id,
+ retry=self.retry,
+ timeout=self.timeout,
+ metadata=self.metadata,
+ )
+ backup = hook.wait_for_operation(operation)
+ self.log.info("Backup %s created successfully", self.backup_id)
+ except HttpError as err:
+ if err.resp.status not in (409, '409'):
+ raise
+ self.log.info("Backup %s already exists", self.backup_id)
+ backup = hook.get_backup(
+ project_id=self.project_id,
+ region=self.region,
+ service_id=self.service_id,
+ backup_id=self.backup_id,
+ retry=self.retry,
+ timeout=self.timeout,
+ metadata=self.metadata,
+ )
+ return Backup.to_dict(backup)
+
+
+class DataprocMetastoreCreateMetadataImportOperator(BaseOperator):
+ """
+ Creates a new MetadataImport in a given project and location.
+
+ :param project_id: Required. The ID of the Google Cloud project that the service belongs to.
+ :type project_id: str
+ :param region: Required. The ID of the Google Cloud region that the service belongs to.
+ :type region: str
+ :param service_id: Required. The ID of the metastore service, which is used as the final component of
+ the metastore service's name. This value must be between 2 and 63 characters long inclusive, begin
+ with a letter, end with a letter or number, and consist of alpha-numeric ASCII characters or
+ hyphens.
+
+ This corresponds to the ``service_id`` field on the ``request`` instance; if ``request`` is
+ provided, this should not be set.
+ :type service_id: str
+ :param metadata_import: Required. The metadata import to create. The ``name`` field is ignored. The ID of
+ the created metadata import must be provided in the request's ``metadata_import_id`` field.
+
+ This corresponds to the ``metadata_import`` field on the ``request`` instance; if ``request`` is
+ provided, this should not be set.
+ :type metadata_import: google.cloud.metastore_v1.types.MetadataImport
+ :param metadata_import_id: Required. The ID of the metadata import, which is used as the final component
+ of the metadata import's name. This value must be between 1 and 64 characters long, begin with a
+ letter, end with a letter or number, and consist of alpha-numeric ASCII characters or hyphens.
+
+ This corresponds to the ``metadata_import_id`` field on the ``request`` instance; if ``request`` is
+ provided, this should not be set.
+ :type metadata_import_id: str
+ :param request_id: Optional. A unique id used to identify the request.
+ :type request_id: str
+ :param retry: Optional. Designation of what errors, if any, should be retried.
+ :type retry: google.api_core.retry.Retry
+ :param timeout: Optional. The timeout for this request.
+ :type timeout: float
+ :param metadata: Optional. Strings which should be sent along with the request as metadata.
+ :type metadata: Sequence[Tuple[str, str]]
+ :param gcp_conn_id: The connection ID to use connecting to Google Cloud.
+ :type gcp_conn_id: str
+ :param impersonation_chain: Optional service account to impersonate using short-term
+ credentials, or chained list of accounts required to get the access_token
+ of the last account in the list, which will be impersonated in the request.
+ If set as a string, the account must grant the originating account
+ the Service Account Token Creator IAM role.
+ If set as a sequence, the identities from the list must grant
+ Service Account Token Creator IAM role to the directly preceding identity, with first
+ account from the list granting this role to the originating account (templated).
+ :type impersonation_chain: Union[str, Sequence[str]]
+ """
+
+ template_fields = (
+ 'project_id',
+ 'metadata_import',
+ 'impersonation_chain',
+ )
+ template_fields_renderers = {'metadata_import': 'json'}
+
+ def __init__(
+ self,
+ *,
+ project_id: str,
+ region: str,
+ service_id: str,
+ metadata_import: MetadataImport,
+ metadata_import_id: str,
+ request_id: Optional[str] = None,
+ retry: Optional[Retry] = None,
+ timeout: Optional[float] = None,
+ metadata: Optional[Sequence[Tuple[str, str]]] = (),
+ gcp_conn_id: str = "google_cloud_default",
+ impersonation_chain: Optional[Union[str, Sequence[str]]] = None,
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.project_id = project_id
+ self.region = region
+ self.service_id = service_id
+ self.metadata_import = metadata_import
+ self.metadata_import_id = metadata_import_id
+ self.request_id = request_id
+ self.retry = retry
+ self.timeout = timeout
+ self.metadata = metadata
+ self.gcp_conn_id = gcp_conn_id
+ self.impersonation_chain = impersonation_chain
+
+ def execute(self, context: dict):
+ hook = DataprocMetastoreHook(
+ gcp_conn_id=self.gcp_conn_id, impersonation_chain=self.impersonation_chain
+ )
+ self.log.info("Creating Dataproc Metastore metadata import: %s", self.metadata_import_id)
+ operation = hook.create_metadata_import(
+ project_id=self.project_id,
+ region=self.region,
+ service_id=self.service_id,
+ metadata_import=self.metadata_import,
+ metadata_import_id=self.metadata_import_id,
+ request_id=self.request_id,
+ retry=self.retry,
+ timeout=self.timeout,
+ metadata=self.metadata,
+ )
+ metadata_import = hook.wait_for_operation(operation)
+ self.log.info("Metadata import %s created successfully", self.metadata_import_id)
+ return metadata_import
+
+
+class DataprocMetastoreCreateServiceOperator(BaseOperator):
+ """
+ Creates a metastore service in a project and location.
+
+ :param region: Required. The ID of the Google Cloud region that the service belongs to.
+ :type region: str
+ :param project_id: Required. The ID of the Google Cloud project that the service belongs to.
+ :type project_id: str
+ :param service: Required. The Metastore service to create. The ``name`` field is ignored. The ID of
+ the created metastore service must be provided in the request's ``service_id`` field.
+
+ This corresponds to the ``service`` field on the ``request`` instance; if ``request`` is provided,
+ this should not be set.
+ :type service: google.cloud.metastore_v1.types.Service
+ :param service_id: Required. The ID of the metastore service, which is used as the final component of
+ the metastore service's name. This value must be between 2 and 63 characters long inclusive, begin
+ with a letter, end with a letter or number, and consist of alpha-numeric ASCII characters or
+ hyphens.
+
+ This corresponds to the ``service_id`` field on the ``request`` instance; if ``request`` is
+ provided, this should not be set.
+ :type service_id: str
+ :param request_id: Optional. A unique id used to identify the request.
+ :type request_id: str
+ :param retry: Designation of what errors, if any, should be retried.
+ :type retry: google.api_core.retry.Retry
+ :param timeout: The timeout for this request.
+ :type timeout: float
+ :param metadata: Strings which should be sent along with the request as metadata.
+ :type metadata: Sequence[Tuple[str, str]]
+ :param gcp_conn_id: The connection ID to use connecting to Google Cloud.
+ :type gcp_conn_id: str
+ :param impersonation_chain: Optional service account to impersonate using short-term
+ credentials, or chained list of accounts required to get the access_token
+ of the last account in the list, which will be impersonated in the request.
+ If set as a string, the account must grant the originating account
+ the Service Account Token Creator IAM role.
+ If set as a sequence, the identities from the list must grant
+ Service Account Token Creator IAM role to the directly preceding identity, with first
+ account from the list granting this role to the originating account (templated).
+ :type impersonation_chain: Union[str, Sequence[str]]
+ """
+
+ template_fields = (
+ 'project_id',
+ 'service',
+ 'impersonation_chain',
+ )
+ template_fields_renderers = {'service': 'json'}
+
+ def __init__(
+ self,
+ *,
+ region: str,
+ project_id: str,
+ service: Optional[Union[Dict, Service]] = None,
+ service_id: str,
+ request_id: Optional[str] = None,
+ retry: Optional[Retry] = None,
+ timeout: Optional[float] = None,
+ metadata: Optional[Sequence[Tuple[str, str]]] = (),
+ gcp_conn_id: str = "google_cloud_default",
+ impersonation_chain: Optional[Union[str, Sequence[str]]] = None,
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.region = region
+ self.project_id = project_id
+ self.service = service
+ self.service_id = service_id
+ self.request_id = request_id
+ self.retry = retry
+ self.timeout = timeout
+ self.metadata = metadata
+ self.gcp_conn_id = gcp_conn_id
+ self.impersonation_chain = impersonation_chain
+
+ def execute(self, context) -> dict:
+ hook = DataprocMetastoreHook(
+ gcp_conn_id=self.gcp_conn_id, impersonation_chain=self.impersonation_chain
+ )
+ self.log.info("Creating Dataproc Metastore service: %s", self.project_id)
+ try:
+ operation = hook.create_service(
+ region=self.region,
+ project_id=self.project_id,
+ service=self.service,
+ service_id=self.service_id,
+ request_id=self.request_id,
+ retry=self.retry,
+ timeout=self.timeout,
+ metadata=self.metadata,
+ )
+ service = hook.wait_for_operation(operation)
+ self.log.info("Service %s created successfully", self.service_id)
+ except HttpError as err:
+ if err.resp.status not in (409, '409'):
+ raise
+ self.log.info("Instance %s already exists", self.service_id)
+ service = hook.get_service(
+ region=self.region,
+ project_id=self.project_id,
+ service_id=self.service_id,
+ retry=self.retry,
+ timeout=self.timeout,
+ metadata=self.metadata,
+ )
+ return Service.to_dict(service)
+
+
+class DataprocMetastoreDeleteBackupOperator(BaseOperator):
+ """
+ Deletes a single backup.
+
+ :param project_id: Required. The ID of the Google Cloud project that the backup belongs to.
+ :type project_id: str
+ :param region: Required. The ID of the Google Cloud region that the backup belongs to.
+ :type region: str
+ :param service_id: Required. The ID of the metastore service, which is used as the final component of
+ the metastore service's name. This value must be between 2 and 63 characters long inclusive, begin
+ with a letter, end with a letter or number, and consist of alpha-numeric ASCII characters or
+ hyphens.
+
+ This corresponds to the ``service_id`` field on the ``request`` instance; if ``request`` is
+ provided, this should not be set.
+ :type service_id: str
+ :param backup_id: Required. The ID of the backup, which is used as the final component of the backup's
+ name. This value must be between 1 and 64 characters long, begin with a letter, end with a letter or
+ number, and consist of alpha-numeric ASCII characters or hyphens.
+
+ This corresponds to the ``backup_id`` field on the ``request`` instance; if ``request`` is provided,
+ this should not be set.
+ :type backup_id: str
+ :param request_id: Optional. A unique id used to identify the request.
+ :type request_id: str
+ :param retry: Optional. Designation of what errors, if any, should be retried.
+ :type retry: google.api_core.retry.Retry
+ :param timeout: Optional. The timeout for this request.
+ :type timeout: float
+ :param metadata: Optional. Strings which should be sent along with the request as metadata.
+ :type metadata: Sequence[Tuple[str, str]]
+ :param gcp_conn_id: The connection ID to use connecting to Google Cloud.
+ :type gcp_conn_id: str
+ :param impersonation_chain: Optional service account to impersonate using short-term
+ credentials, or chained list of accounts required to get the access_token
+ of the last account in the list, which will be impersonated in the request.
+ If set as a string, the account must grant the originating account
+ the Service Account Token Creator IAM role.
+ If set as a sequence, the identities from the list must grant
+ Service Account Token Creator IAM role to the directly preceding identity, with first
+ account from the list granting this role to the originating account (templated).
+ :type impersonation_chain: Union[str, Sequence[str]]
+ """
+
+ template_fields = (
+ 'project_id',
+ 'impersonation_chain',
+ )
+
+ def __init__(
+ self,
+ *,
+ project_id: str,
+ region: str,
+ service_id: str,
+ backup_id: str,
+ request_id: Optional[str] = None,
+ retry: Optional[Retry] = None,
+ timeout: Optional[float] = None,
+ metadata: Optional[Sequence[Tuple[str, str]]] = (),
+ gcp_conn_id: str = "google_cloud_default",
+ impersonation_chain: Optional[Union[str, Sequence[str]]] = None,
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.project_id = project_id
+ self.region = region
+ self.service_id = service_id
+ self.backup_id = backup_id
+ self.request_id = request_id
+ self.retry = retry
+ self.timeout = timeout
+ self.metadata = metadata
+ self.gcp_conn_id = gcp_conn_id
+ self.impersonation_chain = impersonation_chain
+
+ def execute(self, context: dict) -> None:
+ hook = DataprocMetastoreHook(
+ gcp_conn_id=self.gcp_conn_id, impersonation_chain=self.impersonation_chain
+ )
+ self.log.info("Deleting Dataproc Metastore backup: %s", self.backup_id)
+ operation = hook.delete_backup(
+ project_id=self.project_id,
+ region=self.region,
+ service_id=self.service_id,
+ backup_id=self.backup_id,
+ request_id=self.request_id,
+ retry=self.retry,
+ timeout=self.timeout,
+ metadata=self.metadata,
+ )
+ hook.wait_for_operation(operation)
+ self.log.info("Backup %s deleted successfully", self.project_id)
+
+
+class DataprocMetastoreDeleteServiceOperator(BaseOperator):
+ """
+ Deletes a single service.
+
+ :param request: The request object. Request message for
+ [DataprocMetastore.DeleteService][google.cloud.metastore.v1.DataprocMetastore.DeleteService].
+ :type request: google.cloud.metastore_v1.types.DeleteServiceRequest
+ :param project_id: TODO: Fill description
Review comment:
```suggestion
:param project_id: Required. The ID of the Google Cloud project that the service belongs to.
```
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] josh-fell commented on a change in pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
josh-fell commented on a change in pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#discussion_r728360759
##########
File path: airflow/providers/google/cloud/example_dags/example_dataproc_metastore.py
##########
@@ -0,0 +1,202 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example Airflow DAG that show how to use various Dataproc Metastore
+operators to manage a service.
+"""
+
+import os
+
+from airflow import models
+from airflow.providers.google.cloud.operators.dataproc_metastore import (
+ DataprocMetastoreCreateBackupOperator,
+ DataprocMetastoreCreateMetadataImportOperator,
+ DataprocMetastoreCreateServiceOperator,
+ DataprocMetastoreDeleteBackupOperator,
+ DataprocMetastoreDeleteServiceOperator,
+ DataprocMetastoreExportMetadataOperator,
+ DataprocMetastoreGetServiceOperator,
+ DataprocMetastoreListBackupsOperator,
+ DataprocMetastoreRestoreServiceOperator,
+ DataprocMetastoreUpdateServiceOperator,
+)
+from airflow.utils.dates import days_ago
+
+PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "<PROJECT_ID>")
+SERVICE_ID = os.environ.get("GCP_DATAPROC_METASTORE_SERVICE_ID", "dataproc-metastore-system-tests-service-1")
+BACKUP_ID = os.environ.get("GCP_DATAPROC_METASTORE_BACKUP_ID", "dataproc-metastore-system-tests-backup-1")
+REGION = os.environ.get("GCP_REGION", "<REGION>")
+BUCKET = os.environ.get("GCP_DATAPROC_METASTORE_BUCKET", "dataproc-metastore-system-tests")
+METADATA_IMPORT_FILE = os.environ.get("GCS_METADATA_IMPORT_FILE", None)
+METADATA_IMPORT_ID = "dataproc-metastore-system-tests-metadata-import-1"
+TIMEOUT = 1200
+GCS_URI = f"gs://{BUCKET}/{METADATA_IMPORT_FILE}"
+DB_TYPE = "MYSQL"
+DESTINATION_GCS_FOLDER = "gs://{BUCKET}>"
+
+# Service definition
+# Docs: https://cloud.google.com/dataproc-metastore/docs/reference/rest/v1/projects.locations.services#Service
+# [START how_to_cloud_dataproc_metastore_create_service]
+SERVICE = {
+ "name": "test-service",
+}
+# [END how_to_cloud_dataproc_metastore_create_service]
+
+# Update service
+# [START how_to_cloud_dataproc_metastore_update_service]
+SERVICE_TO_UPDATE = {
+ "labels": {
+ "mylocalmachine": "mylocalmachine",
+ "systemtest": "systemtest",
+ }
+}
+UPDATE_MASK = {"paths": ["labels"]}
+# [END how_to_cloud_dataproc_metastore_update_service]
+
+# Backup definition
+# [START how_to_cloud_dataproc_metastore_create_backup]
+BACKUP = {
+ "name": "test-backup",
+}
+# [END how_to_cloud_dataproc_metastore_create_backup]
+
+# Metadata import definition
+# [START how_to_cloud_dataproc_metastore_create_metadata_import]
+METADATA_IMPORT = {
+ "name": "test-metadata-import",
+ "database_dump": {
+ "gcs_uri": GCS_URI,
+ "database_type": DB_TYPE,
+ },
+}
+# [END how_to_cloud_dataproc_metastore_create_metadata_import]
+
+
+with models.DAG("example_gcp_dataproc_metastore", start_date=days_ago(1), schedule_interval="@once") as dag:
+ # [START how_to_cloud_dataproc_metastore_create_service_operator]
+ create_service = DataprocMetastoreCreateServiceOperator(
+ task_id="create_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service=SERVICE,
+ service_id=SERVICE_ID,
+ timeout=TIMEOUT,
+ )
+ # [END how_to_cloud_dataproc_metastore_create_service_operator]
+
+ # [START how_to_cloud_dataproc_metastore_get_service_operator]
+ get_service_details = DataprocMetastoreGetServiceOperator(
+ task_id="get_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service_id=SERVICE_ID,
Review comment:
@mik-laj ❤️ the system tests - the Google provider is absolutely impressive - but adding the `default_args` shouldn't break the system test right? The DAG should still execute successfully unless I missing something in the automated testing. Or is some of the documentation/test cases automated as well and not having the full operator definition in the example DAG breaks the automation?
I'm thinking the variables that are imported for the system test in the DAG could still be defined where they are (so they remain accessible via import) and add `default_args={"region": REGION, "project_id": PROJECT_ID, "service_id": SERVICE_ID}` in the DAG constructor + the explicit code blocks in the operator docs.
> Use default_args makes sense in the examples DAG in the core - airflow/example_dags. The examples in airflow/providers should first of all show how to use operators, not fancy features.
Yes I agree. The provider example DAGs should certainly show how to use the operators and ideally in some useful context/use case but should also demo a real-world way of how the DAG _could_ be authored as well. I'll happily defer to whomever is maintaining the providers for this though. Just thought it was an opportunity for some legitimate use around `default_args` and document 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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mik-laj commented on a change in pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#discussion_r728243407
##########
File path: airflow/providers/google/cloud/example_dags/example_dataproc_metastore.py
##########
@@ -0,0 +1,202 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example Airflow DAG that show how to use various Dataproc Metastore
+operators to manage a service.
+"""
+
+import os
+
+from airflow import models
+from airflow.providers.google.cloud.operators.dataproc_metastore import (
+ DataprocMetastoreCreateBackupOperator,
+ DataprocMetastoreCreateMetadataImportOperator,
+ DataprocMetastoreCreateServiceOperator,
+ DataprocMetastoreDeleteBackupOperator,
+ DataprocMetastoreDeleteServiceOperator,
+ DataprocMetastoreExportMetadataOperator,
+ DataprocMetastoreGetServiceOperator,
+ DataprocMetastoreListBackupsOperator,
+ DataprocMetastoreRestoreServiceOperator,
+ DataprocMetastoreUpdateServiceOperator,
+)
+from airflow.utils.dates import days_ago
+
+PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "<PROJECT_ID>")
+SERVICE_ID = os.environ.get("GCP_DATAPROC_METASTORE_SERVICE_ID", "dataproc-metastore-system-tests-service-1")
+BACKUP_ID = os.environ.get("GCP_DATAPROC_METASTORE_BACKUP_ID", "dataproc-metastore-system-tests-backup-1")
+REGION = os.environ.get("GCP_REGION", "<REGION>")
+BUCKET = os.environ.get("GCP_DATAPROC_METASTORE_BUCKET", "dataproc-metastore-system-tests")
+METADATA_IMPORT_FILE = os.environ.get("GCS_METADATA_IMPORT_FILE", None)
+METADATA_IMPORT_ID = "dataproc-metastore-system-tests-metadata-import-1"
+TIMEOUT = 1200
+GCS_URI = f"gs://{BUCKET}/{METADATA_IMPORT_FILE}"
+DB_TYPE = "MYSQL"
+DESTINATION_GCS_FOLDER = "gs://{BUCKET}>"
+
+# Service definition
+# Docs: https://cloud.google.com/dataproc-metastore/docs/reference/rest/v1/projects.locations.services#Service
+# [START how_to_cloud_dataproc_metastore_create_service]
+SERVICE = {
+ "name": "test-service",
+}
+# [END how_to_cloud_dataproc_metastore_create_service]
+
+# Update service
+# [START how_to_cloud_dataproc_metastore_update_service]
+SERVICE_TO_UPDATE = {
+ "labels": {
+ "mylocalmachine": "mylocalmachine",
+ "systemtest": "systemtest",
+ }
+}
+UPDATE_MASK = {"paths": ["labels"]}
+# [END how_to_cloud_dataproc_metastore_update_service]
+
+# Backup definition
+# [START how_to_cloud_dataproc_metastore_create_backup]
+BACKUP = {
+ "name": "test-backup",
+}
+# [END how_to_cloud_dataproc_metastore_create_backup]
+
+# Metadata import definition
+# [START how_to_cloud_dataproc_metastore_create_metadata_import]
+METADATA_IMPORT = {
+ "name": "test-metadata-import",
+ "database_dump": {
+ "gcs_uri": GCS_URI,
+ "database_type": DB_TYPE,
+ },
+}
+# [END how_to_cloud_dataproc_metastore_create_metadata_import]
+
+
+with models.DAG("example_gcp_dataproc_metastore", start_date=days_ago(1), schedule_interval="@once") as dag:
+ # [START how_to_cloud_dataproc_metastore_create_service_operator]
+ create_service = DataprocMetastoreCreateServiceOperator(
+ task_id="create_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service=SERVICE,
+ service_id=SERVICE_ID,
+ timeout=TIMEOUT,
+ )
+ # [END how_to_cloud_dataproc_metastore_create_service_operator]
+
+ # [START how_to_cloud_dataproc_metastore_get_service_operator]
+ get_service_details = DataprocMetastoreGetServiceOperator(
+ task_id="get_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service_id=SERVICE_ID,
Review comment:
Adding function arguments is not a good idea in this case as it makes it difficult to understand the examples in the generated documentation. The guides are written in such a way that everyone can copy the fragment they need. If we use default_args, copying will be difficult.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] lwyszomi commented on pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
lwyszomi commented on pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#issuecomment-958980408
@mik-laj @josh-fell @SamWheating finally all checks passed, is there a possibility to merge new operators?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk merged pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
potiuk merged pull request #18945:
URL: https://github.com/apache/airflow/pull/18945
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mik-laj commented on a change in pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#discussion_r728243407
##########
File path: airflow/providers/google/cloud/example_dags/example_dataproc_metastore.py
##########
@@ -0,0 +1,202 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example Airflow DAG that show how to use various Dataproc Metastore
+operators to manage a service.
+"""
+
+import os
+
+from airflow import models
+from airflow.providers.google.cloud.operators.dataproc_metastore import (
+ DataprocMetastoreCreateBackupOperator,
+ DataprocMetastoreCreateMetadataImportOperator,
+ DataprocMetastoreCreateServiceOperator,
+ DataprocMetastoreDeleteBackupOperator,
+ DataprocMetastoreDeleteServiceOperator,
+ DataprocMetastoreExportMetadataOperator,
+ DataprocMetastoreGetServiceOperator,
+ DataprocMetastoreListBackupsOperator,
+ DataprocMetastoreRestoreServiceOperator,
+ DataprocMetastoreUpdateServiceOperator,
+)
+from airflow.utils.dates import days_ago
+
+PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "<PROJECT_ID>")
+SERVICE_ID = os.environ.get("GCP_DATAPROC_METASTORE_SERVICE_ID", "dataproc-metastore-system-tests-service-1")
+BACKUP_ID = os.environ.get("GCP_DATAPROC_METASTORE_BACKUP_ID", "dataproc-metastore-system-tests-backup-1")
+REGION = os.environ.get("GCP_REGION", "<REGION>")
+BUCKET = os.environ.get("GCP_DATAPROC_METASTORE_BUCKET", "dataproc-metastore-system-tests")
+METADATA_IMPORT_FILE = os.environ.get("GCS_METADATA_IMPORT_FILE", None)
+METADATA_IMPORT_ID = "dataproc-metastore-system-tests-metadata-import-1"
+TIMEOUT = 1200
+GCS_URI = f"gs://{BUCKET}/{METADATA_IMPORT_FILE}"
+DB_TYPE = "MYSQL"
+DESTINATION_GCS_FOLDER = "gs://{BUCKET}>"
+
+# Service definition
+# Docs: https://cloud.google.com/dataproc-metastore/docs/reference/rest/v1/projects.locations.services#Service
+# [START how_to_cloud_dataproc_metastore_create_service]
+SERVICE = {
+ "name": "test-service",
+}
+# [END how_to_cloud_dataproc_metastore_create_service]
+
+# Update service
+# [START how_to_cloud_dataproc_metastore_update_service]
+SERVICE_TO_UPDATE = {
+ "labels": {
+ "mylocalmachine": "mylocalmachine",
+ "systemtest": "systemtest",
+ }
+}
+UPDATE_MASK = {"paths": ["labels"]}
+# [END how_to_cloud_dataproc_metastore_update_service]
+
+# Backup definition
+# [START how_to_cloud_dataproc_metastore_create_backup]
+BACKUP = {
+ "name": "test-backup",
+}
+# [END how_to_cloud_dataproc_metastore_create_backup]
+
+# Metadata import definition
+# [START how_to_cloud_dataproc_metastore_create_metadata_import]
+METADATA_IMPORT = {
+ "name": "test-metadata-import",
+ "database_dump": {
+ "gcs_uri": GCS_URI,
+ "database_type": DB_TYPE,
+ },
+}
+# [END how_to_cloud_dataproc_metastore_create_metadata_import]
+
+
+with models.DAG("example_gcp_dataproc_metastore", start_date=days_ago(1), schedule_interval="@once") as dag:
+ # [START how_to_cloud_dataproc_metastore_create_service_operator]
+ create_service = DataprocMetastoreCreateServiceOperator(
+ task_id="create_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service=SERVICE,
+ service_id=SERVICE_ID,
+ timeout=TIMEOUT,
+ )
+ # [END how_to_cloud_dataproc_metastore_create_service_operator]
+
+ # [START how_to_cloud_dataproc_metastore_get_service_operator]
+ get_service_details = DataprocMetastoreGetServiceOperator(
+ task_id="get_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service_id=SERVICE_ID,
Review comment:
Adding function arguments is not a good idea in this case as it makes it difficult to understand the examples in the generated documentation. The guides are written in such a way that everyone can copy the fragment they need. If we use default_args, copying will be difficult.
##########
File path: airflow/providers/google/cloud/example_dags/example_dataproc_metastore.py
##########
@@ -0,0 +1,202 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example Airflow DAG that show how to use various Dataproc Metastore
+operators to manage a service.
+"""
+
+import os
+
+from airflow import models
+from airflow.providers.google.cloud.operators.dataproc_metastore import (
+ DataprocMetastoreCreateBackupOperator,
+ DataprocMetastoreCreateMetadataImportOperator,
+ DataprocMetastoreCreateServiceOperator,
+ DataprocMetastoreDeleteBackupOperator,
+ DataprocMetastoreDeleteServiceOperator,
+ DataprocMetastoreExportMetadataOperator,
+ DataprocMetastoreGetServiceOperator,
+ DataprocMetastoreListBackupsOperator,
+ DataprocMetastoreRestoreServiceOperator,
+ DataprocMetastoreUpdateServiceOperator,
+)
+from airflow.utils.dates import days_ago
+
+PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "<PROJECT_ID>")
+SERVICE_ID = os.environ.get("GCP_DATAPROC_METASTORE_SERVICE_ID", "dataproc-metastore-system-tests-service-1")
+BACKUP_ID = os.environ.get("GCP_DATAPROC_METASTORE_BACKUP_ID", "dataproc-metastore-system-tests-backup-1")
+REGION = os.environ.get("GCP_REGION", "<REGION>")
+BUCKET = os.environ.get("GCP_DATAPROC_METASTORE_BUCKET", "dataproc-metastore-system-tests")
Review comment:
For security reasons, we should use bucket names that are invalid and impossible to create, e.g. contain special characters.
##########
File path: airflow/providers/google/cloud/hooks/dataproc_metastore.py
##########
@@ -0,0 +1,673 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+"""This module contains a Google Cloud Dataproc Metastore hook."""
+
+from time import sleep
+from typing import Dict, Optional, Sequence, Tuple, Union
+
+from google.api_core.operation import Operation
+from google.api_core.retry import Retry, exponential_sleep_generator
+from google.cloud.metastore_v1 import DataprocMetastoreClient
+from google.cloud.metastore_v1.types import Backup, MetadataImport, Service
+from google.cloud.metastore_v1.types.metastore import DatabaseDumpSpec, Restore
+from google.protobuf.field_mask_pb2 import FieldMask
+
+from airflow.exceptions import AirflowException
+from airflow.providers.google.common.hooks.base_google import GoogleBaseHook
+
+
+class DataprocMetastoreHook(GoogleBaseHook):
+ """Hook for Google Cloud Dataproc Metastore APIs."""
+
+ def get_dataproc_metastore_client(self, region: Optional[str] = None) -> DataprocMetastoreClient:
+ """Returns DataprocMetastoreClient."""
+ client_options = None
+ if region and region != 'global':
+ client_options = {'api_endpoint': f'{region}-metastore.googleapis.com:443'}
+
+ return DataprocMetastoreClient(
+ credentials=self._get_credentials(), client_info=self.client_info, client_options=client_options
+ )
+
+ def wait_for_operation(self, operation: Operation):
+ """Waits for long-lasting operation to complete."""
+ for time_to_wait in exponential_sleep_generator(initial=10, maximum=120):
+ sleep(time_to_wait)
+ if operation.done():
+ break
+
+ error = operation.exception()
+ if error:
+ raise AirflowException(error)
+ return operation.result()
+
+ def create_backup(
+ self,
+ project_id: str,
+ region: str,
+ service_id: str,
+ backup: Backup,
+ backup_id: str,
+ request_id: Optional[str] = None,
+ retry: Optional[Retry] = None,
+ timeout: Optional[float] = None,
+ metadata: Optional[Sequence[Tuple[str, str]]] = None,
+ ):
+ """
+ Creates a new backup in a given project and location.
+
+ :param project_id: Required. The ID of the Google Cloud project that the service belongs to.
+ :type project_id: str
+ :param region: Required. The ID of the Google Cloud region that the service belongs to.
+ :type region: str
+ :param service_id: Required. The ID of the metastore service, which is used as the final component of
+ the metastore service's name. This value must be between 2 and 63 characters long inclusive, begin
+ with a letter, end with a letter or number, and consist of alpha-numeric ASCII characters or
+ hyphens.
+
+ This corresponds to the ``service_id`` field on the ``request`` instance; if ``request`` is
+ provided, this should not be set.
+ :type service_id: str
+ :param backup: Required. The backup to create. The ``name`` field is ignored. The ID of the created
+ backup must be provided in the request's ``backup_id`` field.
+
+ This corresponds to the ``backup`` field on the ``request`` instance; if ``request`` is provided,
+ this should not be set.
+ :type backup: google.cloud.metastore_v1.types.Backup
+ :param backup_id: Required. The ID of the backup, which is used as the final component of the
+ backup's name. This value must be between 1 and 64 characters long, begin with a letter, end with
+ a letter or number, and consist of alpha-numeric ASCII characters or hyphens.
+
+ This corresponds to the ``backup_id`` field on the ``request`` instance; if ``request`` is
+ provided, this should not be set.
+ :type backup_id: str
+ :param request_id: Optional. A unique id used to identify the request.
+ :type request_id: str
+ :param retry: Designation of what errors, if any, should be retried.
+ :type retry: google.api_core.retry.Retry
+ :param timeout: The timeout for this request.
+ :type timeout: float
+ :param metadata: Strings which should be sent along with the request as metadata.
+ :type metadata: Sequence[Tuple[str, str]]
+ """
+ parent = f'projects/{project_id}/locations/{region}/services/{service_id}'
+
+ client = self.get_dataproc_metastore_client(region=region)
+ result = client.create_backup(
+ request={
+ 'parent': parent,
+ 'backup': backup,
+ 'backup_id': backup_id,
+ 'request_id': request_id,
+ },
+ retry=retry,
+ timeout=timeout,
+ metadata=metadata,
+ )
+ return result
+
+ def create_metadata_import(
Review comment:
Can you add `@GoogleBaseHook.fallback_to_default_project_id` decorator?
##########
File path: airflow/providers/google/cloud/hooks/dataproc_metastore.py
##########
@@ -0,0 +1,673 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+"""This module contains a Google Cloud Dataproc Metastore hook."""
+
+from time import sleep
+from typing import Dict, Optional, Sequence, Tuple, Union
+
+from google.api_core.operation import Operation
+from google.api_core.retry import Retry, exponential_sleep_generator
+from google.cloud.metastore_v1 import DataprocMetastoreClient
+from google.cloud.metastore_v1.types import Backup, MetadataImport, Service
+from google.cloud.metastore_v1.types.metastore import DatabaseDumpSpec, Restore
+from google.protobuf.field_mask_pb2 import FieldMask
+
+from airflow.exceptions import AirflowException
+from airflow.providers.google.common.hooks.base_google import GoogleBaseHook
+
+
+class DataprocMetastoreHook(GoogleBaseHook):
+ """Hook for Google Cloud Dataproc Metastore APIs."""
+
+ def get_dataproc_metastore_client(self, region: Optional[str] = None) -> DataprocMetastoreClient:
+ """Returns DataprocMetastoreClient."""
+ client_options = None
+ if region and region != 'global':
+ client_options = {'api_endpoint': f'{region}-metastore.googleapis.com:443'}
+
+ return DataprocMetastoreClient(
+ credentials=self._get_credentials(), client_info=self.client_info, client_options=client_options
+ )
+
+ def wait_for_operation(self, operation: Operation):
+ """Waits for long-lasting operation to complete."""
+ for time_to_wait in exponential_sleep_generator(initial=10, maximum=120):
+ sleep(time_to_wait)
+ if operation.done():
+ break
Review comment:
It is not needed. See: https://github.com/googleapis/python-api-core/blob/e6a3489a79da00a5de919a4a41782cc3b1d7f583/google/api_core/future/polling.py#L95
##########
File path: airflow/providers/google/cloud/example_dags/example_dataproc_metastore.py
##########
@@ -0,0 +1,202 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example Airflow DAG that show how to use various Dataproc Metastore
+operators to manage a service.
+"""
+
+import os
+
+from airflow import models
+from airflow.providers.google.cloud.operators.dataproc_metastore import (
+ DataprocMetastoreCreateBackupOperator,
+ DataprocMetastoreCreateMetadataImportOperator,
+ DataprocMetastoreCreateServiceOperator,
+ DataprocMetastoreDeleteBackupOperator,
+ DataprocMetastoreDeleteServiceOperator,
+ DataprocMetastoreExportMetadataOperator,
+ DataprocMetastoreGetServiceOperator,
+ DataprocMetastoreListBackupsOperator,
+ DataprocMetastoreRestoreServiceOperator,
+ DataprocMetastoreUpdateServiceOperator,
+)
+from airflow.utils.dates import days_ago
+
+PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "<PROJECT_ID>")
+SERVICE_ID = os.environ.get("GCP_DATAPROC_METASTORE_SERVICE_ID", "dataproc-metastore-system-tests-service-1")
+BACKUP_ID = os.environ.get("GCP_DATAPROC_METASTORE_BACKUP_ID", "dataproc-metastore-system-tests-backup-1")
+REGION = os.environ.get("GCP_REGION", "<REGION>")
+BUCKET = os.environ.get("GCP_DATAPROC_METASTORE_BUCKET", "dataproc-metastore-system-tests")
+METADATA_IMPORT_FILE = os.environ.get("GCS_METADATA_IMPORT_FILE", None)
+METADATA_IMPORT_ID = "dataproc-metastore-system-tests-metadata-import-1"
+TIMEOUT = 1200
+GCS_URI = f"gs://{BUCKET}/{METADATA_IMPORT_FILE}"
+DB_TYPE = "MYSQL"
+DESTINATION_GCS_FOLDER = "gs://{BUCKET}>"
+
+# Service definition
+# Docs: https://cloud.google.com/dataproc-metastore/docs/reference/rest/v1/projects.locations.services#Service
+# [START how_to_cloud_dataproc_metastore_create_service]
+SERVICE = {
+ "name": "test-service",
+}
+# [END how_to_cloud_dataproc_metastore_create_service]
+
+# Update service
+# [START how_to_cloud_dataproc_metastore_update_service]
+SERVICE_TO_UPDATE = {
+ "labels": {
+ "mylocalmachine": "mylocalmachine",
+ "systemtest": "systemtest",
+ }
+}
+UPDATE_MASK = {"paths": ["labels"]}
+# [END how_to_cloud_dataproc_metastore_update_service]
+
+# Backup definition
+# [START how_to_cloud_dataproc_metastore_create_backup]
+BACKUP = {
+ "name": "test-backup",
+}
+# [END how_to_cloud_dataproc_metastore_create_backup]
+
+# Metadata import definition
+# [START how_to_cloud_dataproc_metastore_create_metadata_import]
+METADATA_IMPORT = {
+ "name": "test-metadata-import",
+ "database_dump": {
+ "gcs_uri": GCS_URI,
+ "database_type": DB_TYPE,
+ },
+}
+# [END how_to_cloud_dataproc_metastore_create_metadata_import]
+
+
+with models.DAG("example_gcp_dataproc_metastore", start_date=days_ago(1), schedule_interval="@once") as dag:
+ # [START how_to_cloud_dataproc_metastore_create_service_operator]
+ create_service = DataprocMetastoreCreateServiceOperator(
+ task_id="create_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service=SERVICE,
+ service_id=SERVICE_ID,
+ timeout=TIMEOUT,
+ )
+ # [END how_to_cloud_dataproc_metastore_create_service_operator]
+
+ # [START how_to_cloud_dataproc_metastore_get_service_operator]
+ get_service_details = DataprocMetastoreGetServiceOperator(
+ task_id="get_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service_id=SERVICE_ID,
Review comment:
@josh-fell The example DAG files in the `airflow/providers/google` directory are automatically tested so we are sure they are working correctly. If we use blocks of code, it will be difficult for us to provide a similar level of confidence and trust.
This makes sense in the examples DAG in the core - [`airflow/example_dags`](https://github.com/apache/airflow/tree/main/airflow/example_dags). The examples in `airflow/providers` should first of all show how to use operators, not fancy features.
WDYT?
##########
File path: airflow/providers/google/cloud/example_dags/example_dataproc_metastore.py
##########
@@ -0,0 +1,202 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example Airflow DAG that show how to use various Dataproc Metastore
+operators to manage a service.
+"""
+
+import os
+
+from airflow import models
+from airflow.providers.google.cloud.operators.dataproc_metastore import (
+ DataprocMetastoreCreateBackupOperator,
+ DataprocMetastoreCreateMetadataImportOperator,
+ DataprocMetastoreCreateServiceOperator,
+ DataprocMetastoreDeleteBackupOperator,
+ DataprocMetastoreDeleteServiceOperator,
+ DataprocMetastoreExportMetadataOperator,
+ DataprocMetastoreGetServiceOperator,
+ DataprocMetastoreListBackupsOperator,
+ DataprocMetastoreRestoreServiceOperator,
+ DataprocMetastoreUpdateServiceOperator,
+)
+from airflow.utils.dates import days_ago
+
+PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "<PROJECT_ID>")
+SERVICE_ID = os.environ.get("GCP_DATAPROC_METASTORE_SERVICE_ID", "dataproc-metastore-system-tests-service-1")
+BACKUP_ID = os.environ.get("GCP_DATAPROC_METASTORE_BACKUP_ID", "dataproc-metastore-system-tests-backup-1")
+REGION = os.environ.get("GCP_REGION", "<REGION>")
+BUCKET = os.environ.get("GCP_DATAPROC_METASTORE_BUCKET", "dataproc-metastore-system-tests")
+METADATA_IMPORT_FILE = os.environ.get("GCS_METADATA_IMPORT_FILE", None)
+METADATA_IMPORT_ID = "dataproc-metastore-system-tests-metadata-import-1"
+TIMEOUT = 1200
+GCS_URI = f"gs://{BUCKET}/{METADATA_IMPORT_FILE}"
+DB_TYPE = "MYSQL"
+DESTINATION_GCS_FOLDER = "gs://{BUCKET}>"
+
+# Service definition
+# Docs: https://cloud.google.com/dataproc-metastore/docs/reference/rest/v1/projects.locations.services#Service
+# [START how_to_cloud_dataproc_metastore_create_service]
+SERVICE = {
+ "name": "test-service",
+}
+# [END how_to_cloud_dataproc_metastore_create_service]
+
+# Update service
+# [START how_to_cloud_dataproc_metastore_update_service]
+SERVICE_TO_UPDATE = {
+ "labels": {
+ "mylocalmachine": "mylocalmachine",
+ "systemtest": "systemtest",
+ }
+}
+UPDATE_MASK = {"paths": ["labels"]}
+# [END how_to_cloud_dataproc_metastore_update_service]
+
+# Backup definition
+# [START how_to_cloud_dataproc_metastore_create_backup]
+BACKUP = {
+ "name": "test-backup",
+}
+# [END how_to_cloud_dataproc_metastore_create_backup]
+
+# Metadata import definition
+# [START how_to_cloud_dataproc_metastore_create_metadata_import]
+METADATA_IMPORT = {
+ "name": "test-metadata-import",
+ "database_dump": {
+ "gcs_uri": GCS_URI,
+ "database_type": DB_TYPE,
+ },
+}
+# [END how_to_cloud_dataproc_metastore_create_metadata_import]
+
+
+with models.DAG("example_gcp_dataproc_metastore", start_date=days_ago(1), schedule_interval="@once") as dag:
+ # [START how_to_cloud_dataproc_metastore_create_service_operator]
+ create_service = DataprocMetastoreCreateServiceOperator(
+ task_id="create_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service=SERVICE,
+ service_id=SERVICE_ID,
+ timeout=TIMEOUT,
+ )
+ # [END how_to_cloud_dataproc_metastore_create_service_operator]
+
+ # [START how_to_cloud_dataproc_metastore_get_service_operator]
+ get_service_details = DataprocMetastoreGetServiceOperator(
+ task_id="get_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service_id=SERVICE_ID,
Review comment:
@josh-fell The example DAG files in the `airflow/providers/google` directory are automatically tested so we are sure they are working correctly. If we use blocks of code, it will be difficult for us to provide a similar level of confidence and trust. See: https://github.com/apache/airflow/blob/c1982eecad75c68e52a76088eb90ce15b9326436/tests/providers/google/cloud/operators/test_dataproc_metastore_system.py
Use `default_args` makes sense in the examples DAG in the core - [`airflow/example_dags`](https://github.com/apache/airflow/tree/main/airflow/example_dags). The examples in `airflow/providers` should first of all show how to use operators, not fancy features.
WDYT?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mik-laj commented on a change in pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#discussion_r728246580
##########
File path: airflow/providers/google/cloud/hooks/dataproc_metastore.py
##########
@@ -0,0 +1,673 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+"""This module contains a Google Cloud Dataproc Metastore hook."""
+
+from time import sleep
+from typing import Dict, Optional, Sequence, Tuple, Union
+
+from google.api_core.operation import Operation
+from google.api_core.retry import Retry, exponential_sleep_generator
+from google.cloud.metastore_v1 import DataprocMetastoreClient
+from google.cloud.metastore_v1.types import Backup, MetadataImport, Service
+from google.cloud.metastore_v1.types.metastore import DatabaseDumpSpec, Restore
+from google.protobuf.field_mask_pb2 import FieldMask
+
+from airflow.exceptions import AirflowException
+from airflow.providers.google.common.hooks.base_google import GoogleBaseHook
+
+
+class DataprocMetastoreHook(GoogleBaseHook):
+ """Hook for Google Cloud Dataproc Metastore APIs."""
+
+ def get_dataproc_metastore_client(self, region: Optional[str] = None) -> DataprocMetastoreClient:
+ """Returns DataprocMetastoreClient."""
+ client_options = None
+ if region and region != 'global':
+ client_options = {'api_endpoint': f'{region}-metastore.googleapis.com:443'}
+
+ return DataprocMetastoreClient(
+ credentials=self._get_credentials(), client_info=self.client_info, client_options=client_options
+ )
+
+ def wait_for_operation(self, operation: Operation):
+ """Waits for long-lasting operation to complete."""
+ for time_to_wait in exponential_sleep_generator(initial=10, maximum=120):
+ sleep(time_to_wait)
+ if operation.done():
+ break
+
+ error = operation.exception()
+ if error:
+ raise AirflowException(error)
+ return operation.result()
+
+ def create_backup(
+ self,
+ project_id: str,
+ region: str,
+ service_id: str,
+ backup: Backup,
+ backup_id: str,
+ request_id: Optional[str] = None,
+ retry: Optional[Retry] = None,
+ timeout: Optional[float] = None,
+ metadata: Optional[Sequence[Tuple[str, str]]] = None,
+ ):
+ """
+ Creates a new backup in a given project and location.
+
+ :param project_id: Required. The ID of the Google Cloud project that the service belongs to.
+ :type project_id: str
+ :param region: Required. The ID of the Google Cloud region that the service belongs to.
+ :type region: str
+ :param service_id: Required. The ID of the metastore service, which is used as the final component of
+ the metastore service's name. This value must be between 2 and 63 characters long inclusive, begin
+ with a letter, end with a letter or number, and consist of alpha-numeric ASCII characters or
+ hyphens.
+
+ This corresponds to the ``service_id`` field on the ``request`` instance; if ``request`` is
+ provided, this should not be set.
+ :type service_id: str
+ :param backup: Required. The backup to create. The ``name`` field is ignored. The ID of the created
+ backup must be provided in the request's ``backup_id`` field.
+
+ This corresponds to the ``backup`` field on the ``request`` instance; if ``request`` is provided,
+ this should not be set.
+ :type backup: google.cloud.metastore_v1.types.Backup
+ :param backup_id: Required. The ID of the backup, which is used as the final component of the
+ backup's name. This value must be between 1 and 64 characters long, begin with a letter, end with
+ a letter or number, and consist of alpha-numeric ASCII characters or hyphens.
+
+ This corresponds to the ``backup_id`` field on the ``request`` instance; if ``request`` is
+ provided, this should not be set.
+ :type backup_id: str
+ :param request_id: Optional. A unique id used to identify the request.
+ :type request_id: str
+ :param retry: Designation of what errors, if any, should be retried.
+ :type retry: google.api_core.retry.Retry
+ :param timeout: The timeout for this request.
+ :type timeout: float
+ :param metadata: Strings which should be sent along with the request as metadata.
+ :type metadata: Sequence[Tuple[str, str]]
+ """
+ parent = f'projects/{project_id}/locations/{region}/services/{service_id}'
+
+ client = self.get_dataproc_metastore_client(region=region)
+ result = client.create_backup(
+ request={
+ 'parent': parent,
+ 'backup': backup,
+ 'backup_id': backup_id,
+ 'request_id': request_id,
+ },
+ retry=retry,
+ timeout=timeout,
+ metadata=metadata,
+ )
+ return result
+
+ def create_metadata_import(
Review comment:
Can you add `@GoogleBaseHook.fallback_to_default_project_id` decorator?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mik-laj commented on a change in pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#discussion_r728296630
##########
File path: airflow/providers/google/cloud/example_dags/example_dataproc_metastore.py
##########
@@ -0,0 +1,202 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example Airflow DAG that show how to use various Dataproc Metastore
+operators to manage a service.
+"""
+
+import os
+
+from airflow import models
+from airflow.providers.google.cloud.operators.dataproc_metastore import (
+ DataprocMetastoreCreateBackupOperator,
+ DataprocMetastoreCreateMetadataImportOperator,
+ DataprocMetastoreCreateServiceOperator,
+ DataprocMetastoreDeleteBackupOperator,
+ DataprocMetastoreDeleteServiceOperator,
+ DataprocMetastoreExportMetadataOperator,
+ DataprocMetastoreGetServiceOperator,
+ DataprocMetastoreListBackupsOperator,
+ DataprocMetastoreRestoreServiceOperator,
+ DataprocMetastoreUpdateServiceOperator,
+)
+from airflow.utils.dates import days_ago
+
+PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "<PROJECT_ID>")
+SERVICE_ID = os.environ.get("GCP_DATAPROC_METASTORE_SERVICE_ID", "dataproc-metastore-system-tests-service-1")
+BACKUP_ID = os.environ.get("GCP_DATAPROC_METASTORE_BACKUP_ID", "dataproc-metastore-system-tests-backup-1")
+REGION = os.environ.get("GCP_REGION", "<REGION>")
+BUCKET = os.environ.get("GCP_DATAPROC_METASTORE_BUCKET", "dataproc-metastore-system-tests")
+METADATA_IMPORT_FILE = os.environ.get("GCS_METADATA_IMPORT_FILE", None)
+METADATA_IMPORT_ID = "dataproc-metastore-system-tests-metadata-import-1"
+TIMEOUT = 1200
+GCS_URI = f"gs://{BUCKET}/{METADATA_IMPORT_FILE}"
+DB_TYPE = "MYSQL"
+DESTINATION_GCS_FOLDER = "gs://{BUCKET}>"
+
+# Service definition
+# Docs: https://cloud.google.com/dataproc-metastore/docs/reference/rest/v1/projects.locations.services#Service
+# [START how_to_cloud_dataproc_metastore_create_service]
+SERVICE = {
+ "name": "test-service",
+}
+# [END how_to_cloud_dataproc_metastore_create_service]
+
+# Update service
+# [START how_to_cloud_dataproc_metastore_update_service]
+SERVICE_TO_UPDATE = {
+ "labels": {
+ "mylocalmachine": "mylocalmachine",
+ "systemtest": "systemtest",
+ }
+}
+UPDATE_MASK = {"paths": ["labels"]}
+# [END how_to_cloud_dataproc_metastore_update_service]
+
+# Backup definition
+# [START how_to_cloud_dataproc_metastore_create_backup]
+BACKUP = {
+ "name": "test-backup",
+}
+# [END how_to_cloud_dataproc_metastore_create_backup]
+
+# Metadata import definition
+# [START how_to_cloud_dataproc_metastore_create_metadata_import]
+METADATA_IMPORT = {
+ "name": "test-metadata-import",
+ "database_dump": {
+ "gcs_uri": GCS_URI,
+ "database_type": DB_TYPE,
+ },
+}
+# [END how_to_cloud_dataproc_metastore_create_metadata_import]
+
+
+with models.DAG("example_gcp_dataproc_metastore", start_date=days_ago(1), schedule_interval="@once") as dag:
+ # [START how_to_cloud_dataproc_metastore_create_service_operator]
+ create_service = DataprocMetastoreCreateServiceOperator(
+ task_id="create_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service=SERVICE,
+ service_id=SERVICE_ID,
+ timeout=TIMEOUT,
+ )
+ # [END how_to_cloud_dataproc_metastore_create_service_operator]
+
+ # [START how_to_cloud_dataproc_metastore_get_service_operator]
+ get_service_details = DataprocMetastoreGetServiceOperator(
+ task_id="get_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service_id=SERVICE_ID,
Review comment:
@josh-fell The example DAG files in the `airflow/providers/google` directory are automatically tested so we are sure they are working correctly. If we use blocks of code, it will be difficult for us to provide a similar level of confidence and trust.
This makes sense in the examples DAG in the core - [`airflow/example_dags`](https://github.com/apache/airflow/tree/main/airflow/example_dags). The examples in `airflow/providers` should first of all show how to use operators, not fancy features.
WDYT?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] lwyszomi commented on pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
lwyszomi commented on pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#issuecomment-958980408
@mik-laj @josh-fell @SamWheating finally all checks passed, is there a possibility to merge new operators?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mik-laj commented on a change in pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#discussion_r731316602
##########
File path: airflow/providers/google/cloud/hooks/dataproc_metastore.py
##########
@@ -0,0 +1,673 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+"""This module contains a Google Cloud Dataproc Metastore hook."""
+
+from time import sleep
+from typing import Dict, Optional, Sequence, Tuple, Union
+
+from google.api_core.operation import Operation
+from google.api_core.retry import Retry, exponential_sleep_generator
+from google.cloud.metastore_v1 import DataprocMetastoreClient
+from google.cloud.metastore_v1.types import Backup, MetadataImport, Service
+from google.cloud.metastore_v1.types.metastore import DatabaseDumpSpec, Restore
+from google.protobuf.field_mask_pb2 import FieldMask
+
+from airflow.exceptions import AirflowException
+from airflow.providers.google.common.hooks.base_google import GoogleBaseHook
+
+
+class DataprocMetastoreHook(GoogleBaseHook):
+ """Hook for Google Cloud Dataproc Metastore APIs."""
+
+ def get_dataproc_metastore_client(self, region: Optional[str] = None) -> DataprocMetastoreClient:
+ """Returns DataprocMetastoreClient."""
+ client_options = None
+ if region and region != 'global':
+ client_options = {'api_endpoint': f'{region}-metastore.googleapis.com:443'}
+
+ return DataprocMetastoreClient(
+ credentials=self._get_credentials(), client_info=self.client_info, client_options=client_options
+ )
+
+ def wait_for_operation(self, operation: Operation):
+ """Waits for long-lasting operation to complete."""
+ for time_to_wait in exponential_sleep_generator(initial=10, maximum=120):
+ sleep(time_to_wait)
+ if operation.done():
+ break
+
+ error = operation.exception()
+ if error:
+ raise AirflowException(error)
+ return operation.result()
+
+ def create_backup(
+ self,
+ project_id: str,
+ region: str,
+ service_id: str,
+ backup: Backup,
+ backup_id: str,
+ request_id: Optional[str] = None,
+ retry: Optional[Retry] = None,
+ timeout: Optional[float] = None,
+ metadata: Optional[Sequence[Tuple[str, str]]] = None,
+ ):
+ """
+ Creates a new backup in a given project and location.
+
+ :param project_id: Required. The ID of the Google Cloud project that the service belongs to.
+ :type project_id: str
+ :param region: Required. The ID of the Google Cloud region that the service belongs to.
+ :type region: str
+ :param service_id: Required. The ID of the metastore service, which is used as the final component of
+ the metastore service's name. This value must be between 2 and 63 characters long inclusive, begin
+ with a letter, end with a letter or number, and consist of alpha-numeric ASCII characters or
+ hyphens.
+
+ This corresponds to the ``service_id`` field on the ``request`` instance; if ``request`` is
+ provided, this should not be set.
+ :type service_id: str
+ :param backup: Required. The backup to create. The ``name`` field is ignored. The ID of the created
+ backup must be provided in the request's ``backup_id`` field.
+
+ This corresponds to the ``backup`` field on the ``request`` instance; if ``request`` is provided,
+ this should not be set.
+ :type backup: google.cloud.metastore_v1.types.Backup
+ :param backup_id: Required. The ID of the backup, which is used as the final component of the
+ backup's name. This value must be between 1 and 64 characters long, begin with a letter, end with
+ a letter or number, and consist of alpha-numeric ASCII characters or hyphens.
+
+ This corresponds to the ``backup_id`` field on the ``request`` instance; if ``request`` is
+ provided, this should not be set.
+ :type backup_id: str
+ :param request_id: Optional. A unique id used to identify the request.
+ :type request_id: str
+ :param retry: Designation of what errors, if any, should be retried.
+ :type retry: google.api_core.retry.Retry
+ :param timeout: The timeout for this request.
+ :type timeout: float
+ :param metadata: Strings which should be sent along with the request as metadata.
+ :type metadata: Sequence[Tuple[str, str]]
+ """
+ parent = f'projects/{project_id}/locations/{region}/services/{service_id}'
+
+ client = self.get_dataproc_metastore_client(region=region)
+ result = client.create_backup(
+ request={
+ 'parent': parent,
+ 'backup': backup,
+ 'backup_id': backup_id,
+ 'request_id': request_id,
+ },
+ retry=retry,
+ timeout=timeout,
+ metadata=metadata,
+ )
+ return result
+
+ def create_metadata_import(
Review comment:
All public method of hook that accepts project ID parameter.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mik-laj commented on a change in pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#discussion_r728244832
##########
File path: airflow/providers/google/cloud/example_dags/example_dataproc_metastore.py
##########
@@ -0,0 +1,202 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example Airflow DAG that show how to use various Dataproc Metastore
+operators to manage a service.
+"""
+
+import os
+
+from airflow import models
+from airflow.providers.google.cloud.operators.dataproc_metastore import (
+ DataprocMetastoreCreateBackupOperator,
+ DataprocMetastoreCreateMetadataImportOperator,
+ DataprocMetastoreCreateServiceOperator,
+ DataprocMetastoreDeleteBackupOperator,
+ DataprocMetastoreDeleteServiceOperator,
+ DataprocMetastoreExportMetadataOperator,
+ DataprocMetastoreGetServiceOperator,
+ DataprocMetastoreListBackupsOperator,
+ DataprocMetastoreRestoreServiceOperator,
+ DataprocMetastoreUpdateServiceOperator,
+)
+from airflow.utils.dates import days_ago
+
+PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "<PROJECT_ID>")
+SERVICE_ID = os.environ.get("GCP_DATAPROC_METASTORE_SERVICE_ID", "dataproc-metastore-system-tests-service-1")
+BACKUP_ID = os.environ.get("GCP_DATAPROC_METASTORE_BACKUP_ID", "dataproc-metastore-system-tests-backup-1")
+REGION = os.environ.get("GCP_REGION", "<REGION>")
+BUCKET = os.environ.get("GCP_DATAPROC_METASTORE_BUCKET", "dataproc-metastore-system-tests")
Review comment:
For security reasons, we should use bucket names that are invalid and impossible to create, e.g. contain special characters.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mik-laj commented on a change in pull request #18945: Add dataproc metastore operators
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #18945:
URL: https://github.com/apache/airflow/pull/18945#discussion_r728417941
##########
File path: airflow/providers/google/cloud/example_dags/example_dataproc_metastore.py
##########
@@ -0,0 +1,202 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example Airflow DAG that show how to use various Dataproc Metastore
+operators to manage a service.
+"""
+
+import os
+
+from airflow import models
+from airflow.providers.google.cloud.operators.dataproc_metastore import (
+ DataprocMetastoreCreateBackupOperator,
+ DataprocMetastoreCreateMetadataImportOperator,
+ DataprocMetastoreCreateServiceOperator,
+ DataprocMetastoreDeleteBackupOperator,
+ DataprocMetastoreDeleteServiceOperator,
+ DataprocMetastoreExportMetadataOperator,
+ DataprocMetastoreGetServiceOperator,
+ DataprocMetastoreListBackupsOperator,
+ DataprocMetastoreRestoreServiceOperator,
+ DataprocMetastoreUpdateServiceOperator,
+)
+from airflow.utils.dates import days_ago
+
+PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "<PROJECT_ID>")
+SERVICE_ID = os.environ.get("GCP_DATAPROC_METASTORE_SERVICE_ID", "dataproc-metastore-system-tests-service-1")
+BACKUP_ID = os.environ.get("GCP_DATAPROC_METASTORE_BACKUP_ID", "dataproc-metastore-system-tests-backup-1")
+REGION = os.environ.get("GCP_REGION", "<REGION>")
+BUCKET = os.environ.get("GCP_DATAPROC_METASTORE_BUCKET", "dataproc-metastore-system-tests")
+METADATA_IMPORT_FILE = os.environ.get("GCS_METADATA_IMPORT_FILE", None)
+METADATA_IMPORT_ID = "dataproc-metastore-system-tests-metadata-import-1"
+TIMEOUT = 1200
+GCS_URI = f"gs://{BUCKET}/{METADATA_IMPORT_FILE}"
+DB_TYPE = "MYSQL"
+DESTINATION_GCS_FOLDER = "gs://{BUCKET}>"
+
+# Service definition
+# Docs: https://cloud.google.com/dataproc-metastore/docs/reference/rest/v1/projects.locations.services#Service
+# [START how_to_cloud_dataproc_metastore_create_service]
+SERVICE = {
+ "name": "test-service",
+}
+# [END how_to_cloud_dataproc_metastore_create_service]
+
+# Update service
+# [START how_to_cloud_dataproc_metastore_update_service]
+SERVICE_TO_UPDATE = {
+ "labels": {
+ "mylocalmachine": "mylocalmachine",
+ "systemtest": "systemtest",
+ }
+}
+UPDATE_MASK = {"paths": ["labels"]}
+# [END how_to_cloud_dataproc_metastore_update_service]
+
+# Backup definition
+# [START how_to_cloud_dataproc_metastore_create_backup]
+BACKUP = {
+ "name": "test-backup",
+}
+# [END how_to_cloud_dataproc_metastore_create_backup]
+
+# Metadata import definition
+# [START how_to_cloud_dataproc_metastore_create_metadata_import]
+METADATA_IMPORT = {
+ "name": "test-metadata-import",
+ "database_dump": {
+ "gcs_uri": GCS_URI,
+ "database_type": DB_TYPE,
+ },
+}
+# [END how_to_cloud_dataproc_metastore_create_metadata_import]
+
+
+with models.DAG("example_gcp_dataproc_metastore", start_date=days_ago(1), schedule_interval="@once") as dag:
+ # [START how_to_cloud_dataproc_metastore_create_service_operator]
+ create_service = DataprocMetastoreCreateServiceOperator(
+ task_id="create_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service=SERVICE,
+ service_id=SERVICE_ID,
+ timeout=TIMEOUT,
+ )
+ # [END how_to_cloud_dataproc_metastore_create_service_operator]
+
+ # [START how_to_cloud_dataproc_metastore_get_service_operator]
+ get_service_details = DataprocMetastoreGetServiceOperator(
+ task_id="get_service",
+ region=REGION,
+ project_id=PROJECT_ID,
+ service_id=SERVICE_ID,
Review comment:
This will work as long as the job definition is in a separate Python file. To add examples to documentation, we use `example-include` directive, To allow us to keep docs and example dags in sync
https://github.com/apache/airflow/blob/af4a5e006e4f5c9f203afeac039b22c6adee317f/docs/apache-airflow-providers-google/operators/cloud/dataproc.rst#L54-L58
Alternatively, we can also use the `code-block` directive, but this means that the example will not be tested and executed on CI. If it is not running on a CI, the examples very quickly become out-of-date and are more of a burden to maintain.
https://github.com/apache/airflow/blob/060345c0d982765e39da5fa8b2e2c6a01e89e394/docs/apache-airflow/tutorial.rst#L205-L231
For example:
In [this PR](https://github.com/apache/airflow/pull/18036/files), I deprecated default value of `pod_name` parameter in `EKSCreateClusterOperator`. This meant that if someone loads the original example file, `DeprecationWarning` will be displayed. But we also [detect such problems](https://github.com/apache/airflow/pull/17900) automatically on CI, and we won't allow a merge PR that doesn't update the examples. This makes these examples much easier to maintain. We don't have to worry if all the examples have been updated to show the best practices because we detect all such problems automatically and if something still needs to be changed, we will inform everyone on the CI. All this without the involvement of any of the project maintainers.
If we used the `code-block` directive, we would have to find a way to detect similar problems automatically as well, or the examples would be out of date very quickly.
This is one example, but we have other automated checks that allow us to take care of the high quality of documentation and examples
> Just thought it was an opportunity for some legitimate use around default_args and document it.
I agree that this is a good use for `default_args`, but given the problems with using these examples in the documentation, we don't necessarily have to do in all cases.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org