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/09/12 21:11:17 UTC
[GitHub] [airflow] iostreamdoth opened a new pull request #18197: Issue-17089: add gcs yml config to create dataproc cluster
iostreamdoth opened a new pull request #18197:
URL: https://github.com/apache/airflow/pull/18197
<!--
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: #17089
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
---
**^ Add meaningful description above**
- Add functionality for creating dataproc cluster and workflow config from yml config given on a gcs path.
- Add documentation for creating cluster and workflow cluster configs from yml file in gcs.
- Add Tests for prepare template methods for all 3 operators.
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] SamWheating commented on a change in pull request #18197: Add gcs yml config to create dataproc cluster
Posted by GitBox <gi...@apache.org>.
SamWheating commented on a change in pull request #18197:
URL: https://github.com/apache/airflow/pull/18197#discussion_r711838902
##########
File path: tests/providers/google/cloud/operators/test_dataproc_system.py
##########
@@ -54,12 +75,14 @@ def setUp(self):
self.create_gcs_bucket(BUCKET)
self.upload_content_to_gcs(lines=pyspark_file, bucket=GCS_URI, filename=PYSPARK_MAIN)
self.upload_content_to_gcs(lines=sparkr_file, bucket=GCS_URI, filename=SPARKR_MAIN)
+ self.upload_content_to_gcs(lines=yml_config_file, bucket=GCS_URI, filename=YML_CONFIG_FILE)
Review comment:
Using both `yml_config_file` and `YML_CONFIG_FILE` is bad for readability in my opinion, maybe one should be renamed to reduce any confusion.
--
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] iostreamdoth commented on a change in pull request #18197: Add gcs yml config to create dataproc cluster
Posted by GitBox <gi...@apache.org>.
iostreamdoth commented on a change in pull request #18197:
URL: https://github.com/apache/airflow/pull/18197#discussion_r707831115
##########
File path: airflow/providers/google/cloud/operators/dataproc.py
##########
@@ -48,6 +54,83 @@
)
+class DataProcCreateWorkflowBaseOperator(BaseOperator):
+ """Helper class for preparing configs from workflow template"""
+
+ def __init__(self, **kwargs):
+ super().__init__(**kwargs)
+
+ def prepare_template(self) -> None:
+ try:
+ # if no file is specified, skip
+ if isinstance(self.template["placement"]["managed_cluster"]["config"], str):
+ if urlparse(self.template["placement"]["managed_cluster"]["config"]).scheme == "gs":
+ config = self._build_cluster_config(
+ self.template["placement"]["managed_cluster"]["config"]
+ )
+
+ # Dataproc workflows do not accept life cycle_config
+ if "lifecycle_config" in config:
+ self.log.warning(
+ "Workflow does not accept lifecycle in config, removing it from config"
+ )
+ config.pop("lifecycle_config", None)
+ self.template["placement"]["managed_cluster"]["config"] = config
+ else:
+ self.log.warning(
+ "Workflow Template config accepts only gcs path string or dictioanry type, "
+ "skipping build cluster config from yml file"
+ )
+ except KeyError:
+ pass
Review comment:
A template dictionary can be of 2 types:
1. managed_cluster which needs config key like
```
WORKFLOW_TEMPLATE_YML = {
"id": WORKFLOW_NAME_YML,
"placement": {
"**managed_cluster**": {
"cluster_name": CLUSTER_NAME,
"config": YML_CONFIG_URI,
}
},
"jobs": [{"step_id": "pig_job_1", "pig_job": PIG_JOB["pig_job"]}],
}
```
2. cluster_selector like which does not need
```
WORKFLOW_TEMPLATE_YML = {
"id": WORKFLOW_NAME_YML,
"placement": {
"**cluster_selector** ": {
"cluster_name": CLUSTER_NAME
}
},
"jobs": [{"step_id": "pig_job_1", "pig_job": PIG_JOB["pig_job"]}],
}
```
so given that, yes key `self.template["placement"]["managed_cluster"]["config"]` can fail in case of `cluster_selector` type template but it is still a valid template. I did more digging around it and I feel that prepare_template() is not the right place to perform this conversion of yml config file to dictionary. Will try to resolve the config during execute method.
--
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] turbaszek commented on a change in pull request #18197: Add gcs yml config to create dataproc cluster
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #18197:
URL: https://github.com/apache/airflow/pull/18197#discussion_r707051981
##########
File path: airflow/providers/google/cloud/example_dags/example_dataproc.py
##########
@@ -45,6 +45,8 @@
PYSPARK_URI = f"gs://{BUCKET}/{PYSPARK_MAIN}"
SPARKR_MAIN = os.environ.get("SPARKR_MAIN", "hello_world.R")
SPARKR_URI = f"gs://{BUCKET}/{SPARKR_MAIN}"
+YML_CONFIG_FILE = os.environ.get("YAML_CONFIG_FILE", "config.yml")
Review comment:
The config.yaml is missing so this DAG will not work, correct?
--
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] github-actions[bot] commented on pull request #18197: Add gcs yml config to create dataproc cluster
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #18197:
URL: https://github.com/apache/airflow/pull/18197#issuecomment-962708054
This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.
--
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] iostreamdoth edited a comment on pull request #18197: Add gcs yml config to create dataproc cluster
Posted by GitBox <gi...@apache.org>.
iostreamdoth edited a comment on pull request #18197:
URL: https://github.com/apache/airflow/pull/18197#issuecomment-917711447
https://github.com/apache/airflow/issues/17089 this is the issue discussion for reference
--
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] iostreamdoth commented on a change in pull request #18197: Add gcs yml config to create dataproc cluster
Posted by GitBox <gi...@apache.org>.
iostreamdoth commented on a change in pull request #18197:
URL: https://github.com/apache/airflow/pull/18197#discussion_r707831115
##########
File path: airflow/providers/google/cloud/operators/dataproc.py
##########
@@ -48,6 +54,83 @@
)
+class DataProcCreateWorkflowBaseOperator(BaseOperator):
+ """Helper class for preparing configs from workflow template"""
+
+ def __init__(self, **kwargs):
+ super().__init__(**kwargs)
+
+ def prepare_template(self) -> None:
+ try:
+ # if no file is specified, skip
+ if isinstance(self.template["placement"]["managed_cluster"]["config"], str):
+ if urlparse(self.template["placement"]["managed_cluster"]["config"]).scheme == "gs":
+ config = self._build_cluster_config(
+ self.template["placement"]["managed_cluster"]["config"]
+ )
+
+ # Dataproc workflows do not accept life cycle_config
+ if "lifecycle_config" in config:
+ self.log.warning(
+ "Workflow does not accept lifecycle in config, removing it from config"
+ )
+ config.pop("lifecycle_config", None)
+ self.template["placement"]["managed_cluster"]["config"] = config
+ else:
+ self.log.warning(
+ "Workflow Template config accepts only gcs path string or dictioanry type, "
+ "skipping build cluster config from yml file"
+ )
+ except KeyError:
+ pass
Review comment:
A template dictionary can be of 2 types:
1. managed_cluster which needs config key like
```
WORKFLOW_TEMPLATE_YML = {
"id": WORKFLOW_NAME_YML,
"placement": {
"managed_cluster": {
"cluster_name": CLUSTER_NAME,
"config": YML_CONFIG_URI,
}
},
"jobs": [{"step_id": "pig_job_1", "pig_job": PIG_JOB["pig_job"]}],
}
```
2. cluster_selector like which does not need config key
```
WORKFLOW_TEMPLATE = {
"id": WORKFLOW_NAME
"placement": {
"cluster_selector": {
"cluster_name": CLUSTER_NAME
}
},
"jobs": [{"step_id": "pig_job_1", "pig_job": PIG_JOB["pig_job"]}],
}
```
so given that, yes key `self.template["placement"]["managed_cluster"]["config"]` can fail in case of `cluster_selector` type template but it is still a valid template. I did more digging around it and I feel that prepare_template() is not the right place to perform this conversion of yml config file to dictionary. Will try to resolve the config during execute method.
--
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 #18197: Add gcs yml config to create dataproc cluster
Posted by GitBox <gi...@apache.org>.
SamWheating commented on a change in pull request #18197:
URL: https://github.com/apache/airflow/pull/18197#discussion_r711839969
##########
File path: airflow/providers/google/cloud/operators/dataproc.py
##########
@@ -527,7 +629,7 @@ def __init__(
cluster_name: str,
region: Optional[str] = None,
project_id: Optional[str] = None,
- cluster_config: Optional[Dict] = None,
+ cluster_config: Optional[Union[Dict, str]] = None,
Review comment:
Accepting strings here might cause some issues with rendering templated values when a user wants to use a templated string, due to the above setting:
```
template_fields_renderers = {'cluster_config': 'json'}
```
--
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 #18197: Add gcs yml config to create dataproc cluster
Posted by GitBox <gi...@apache.org>.
SamWheating commented on a change in pull request #18197:
URL: https://github.com/apache/airflow/pull/18197#discussion_r711837965
##########
File path: airflow/providers/google/cloud/operators/dataproc.py
##########
@@ -663,6 +765,16 @@ def execute(self, context) -> dict:
self.log.info('Creating cluster: %s', self.cluster_name)
hook = DataprocHook(gcp_conn_id=self.gcp_conn_id, impersonation_chain=self.impersonation_chain)
# Save data required to display extra link no matter what the cluster status will be
Review comment:
Could you move this comment to keep it next to the `sel.xcom_push()` which it refers 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] iostreamdoth commented on a change in pull request #18197: Add gcs yml config to create dataproc cluster
Posted by GitBox <gi...@apache.org>.
iostreamdoth commented on a change in pull request #18197:
URL: https://github.com/apache/airflow/pull/18197#discussion_r707831115
##########
File path: airflow/providers/google/cloud/operators/dataproc.py
##########
@@ -48,6 +54,83 @@
)
+class DataProcCreateWorkflowBaseOperator(BaseOperator):
+ """Helper class for preparing configs from workflow template"""
+
+ def __init__(self, **kwargs):
+ super().__init__(**kwargs)
+
+ def prepare_template(self) -> None:
+ try:
+ # if no file is specified, skip
+ if isinstance(self.template["placement"]["managed_cluster"]["config"], str):
+ if urlparse(self.template["placement"]["managed_cluster"]["config"]).scheme == "gs":
+ config = self._build_cluster_config(
+ self.template["placement"]["managed_cluster"]["config"]
+ )
+
+ # Dataproc workflows do not accept life cycle_config
+ if "lifecycle_config" in config:
+ self.log.warning(
+ "Workflow does not accept lifecycle in config, removing it from config"
+ )
+ config.pop("lifecycle_config", None)
+ self.template["placement"]["managed_cluster"]["config"] = config
+ else:
+ self.log.warning(
+ "Workflow Template config accepts only gcs path string or dictioanry type, "
+ "skipping build cluster config from yml file"
+ )
+ except KeyError:
+ pass
Review comment:
A template dictionary can be of 2 types:
1. managed_cluster which needs config key like
```
WORKFLOW_TEMPLATE_YML = {
"id": WORKFLOW_NAME_YML,
"placement": {
"**managed_cluster**": {
"cluster_name": CLUSTER_NAME,
"config": YML_CONFIG_URI,
}
},
"jobs": [{"step_id": "pig_job_1", "pig_job": PIG_JOB["pig_job"]}],
}
```
2. cluster_selector like which does not need config key
```
WORKFLOW_TEMPLATE_YML = {
"id": WORKFLOW_NAME_YML,
"placement": {
"**cluster_selector** ": {
"cluster_name": CLUSTER_NAME
}
},
"jobs": [{"step_id": "pig_job_1", "pig_job": PIG_JOB["pig_job"]}],
}
```
so given that, yes key `self.template["placement"]["managed_cluster"]["config"]` can fail in case of `cluster_selector` type template but it is still a valid template. I did more digging around it and I feel that prepare_template() is not the right place to perform this conversion of yml config file to dictionary. Will try to resolve the config during execute method.
--
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] iostreamdoth commented on pull request #18197: Add gcs yml config to create dataproc cluster
Posted by GitBox <gi...@apache.org>.
iostreamdoth commented on pull request #18197:
URL: https://github.com/apache/airflow/pull/18197#issuecomment-917711447
https://github.com/apache/airflow/issues/17089
--
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] iostreamdoth commented on a change in pull request #18197: Add gcs yml config to create dataproc cluster
Posted by GitBox <gi...@apache.org>.
iostreamdoth commented on a change in pull request #18197:
URL: https://github.com/apache/airflow/pull/18197#discussion_r707075616
##########
File path: airflow/providers/google/cloud/operators/dataproc.py
##########
@@ -48,6 +54,83 @@
)
+class DataProcCreateWorkflowBaseOperator(BaseOperator):
+ """Helper class for preparing configs from workflow template"""
+
+ def __init__(self, **kwargs):
+ super().__init__(**kwargs)
+
+ def prepare_template(self) -> None:
+ try:
+ # if no file is specified, skip
+ if isinstance(self.template["placement"]["managed_cluster"]["config"], str):
+ if urlparse(self.template["placement"]["managed_cluster"]["config"]).scheme == "gs":
+ config = self._build_cluster_config(
+ self.template["placement"]["managed_cluster"]["config"]
+ )
+
+ # Dataproc workflows do not accept life cycle_config
+ if "lifecycle_config" in config:
+ self.log.warning(
+ "Workflow does not accept lifecycle in config, removing it from config"
+ )
+ config.pop("lifecycle_config", None)
+ self.template["placement"]["managed_cluster"]["config"] = config
+ else:
+ self.log.warning(
+ "Workflow Template config accepts only gcs path string or dictioanry type, "
+ "skipping build cluster config from yml file"
+ )
+ except KeyError:
+ pass
Review comment:
I believe google-cloud-dataproc takes care of validation of workflow placement message. These are optional either managed_cluster or cluster_selector.
https://googleapis.dev/python/dataproc/latest/dataproc_v1beta2/types.html#google.cloud.dataproc_v1beta2.types.WorkflowTemplatePlacement
--
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 #18197: Add gcs yml config to create dataproc cluster
Posted by GitBox <gi...@apache.org>.
SamWheating commented on a change in pull request #18197:
URL: https://github.com/apache/airflow/pull/18197#discussion_r711836314
##########
File path: airflow/providers/google/cloud/operators/dataproc.py
##########
@@ -48,6 +54,101 @@
)
+class DataProcCreateWorkflowBaseOperator(BaseOperator):
+ """Helper class for preparing configs from workflow template"""
+
+ def __init__(self, **kwargs):
+ super().__init__(**kwargs)
+
+ def _template_has_placement(self):
+ return True if "placement" in self.template else False
+
+ def _template_is_managed_cluster(self):
+
+ if self._template_has_placement():
+ return True if "managed_cluster" in self.template["placement"] else False
+ else:
+ # Placement is required field in the template
+ raise KeyError("Placement is missing")
+
+ def _has_valid_config(self):
+ if self._template_is_managed_cluster():
+ # cluster templates can be of 2 types, managed_cluster or cluster_selector.
+ return True if "config" in self.template["placement"]["managed_cluster"] else False
Review comment:
```suggestion
return "config" in self.template["placement"]["managed_cluster"]
```
There's a lot of places in this file which look like:
```python
return True if <condition> else False
```
Which can all be shortened to just
```python
return <condition>
```
--
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] turbaszek commented on a change in pull request #18197: Add gcs yml config to create dataproc cluster
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #18197:
URL: https://github.com/apache/airflow/pull/18197#discussion_r707056889
##########
File path: airflow/providers/google/cloud/operators/dataproc.py
##########
@@ -48,6 +54,83 @@
)
+class DataProcCreateWorkflowBaseOperator(BaseOperator):
+ """Helper class for preparing configs from workflow template"""
+
+ def __init__(self, **kwargs):
+ super().__init__(**kwargs)
+
+ def prepare_template(self) -> None:
+ try:
+ # if no file is specified, skip
+ if isinstance(self.template["placement"]["managed_cluster"]["config"], str):
+ if urlparse(self.template["placement"]["managed_cluster"]["config"]).scheme == "gs":
+ config = self._build_cluster_config(
+ self.template["placement"]["managed_cluster"]["config"]
+ )
+
+ # Dataproc workflows do not accept life cycle_config
+ if "lifecycle_config" in config:
+ self.log.warning(
+ "Workflow does not accept lifecycle in config, removing it from config"
+ )
+ config.pop("lifecycle_config", None)
+ self.template["placement"]["managed_cluster"]["config"] = config
+ else:
+ self.log.warning(
+ "Workflow Template config accepts only gcs path string or dictioanry type, "
+ "skipping build cluster config from yml file"
+ )
+ except KeyError:
+ pass
Review comment:
Hm, I'm not sure if ignoring any `KeyError` is a good approach. Or do we expect that the template will be wrong and the validation will happen on Dataproc side?
--
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] turbaszek commented on a change in pull request #18197: Add gcs yml config to create dataproc cluster
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #18197:
URL: https://github.com/apache/airflow/pull/18197#discussion_r707358215
##########
File path: airflow/providers/google/cloud/operators/dataproc.py
##########
@@ -48,6 +54,83 @@
)
+class DataProcCreateWorkflowBaseOperator(BaseOperator):
+ """Helper class for preparing configs from workflow template"""
+
+ def __init__(self, **kwargs):
+ super().__init__(**kwargs)
+
+ def prepare_template(self) -> None:
+ try:
+ # if no file is specified, skip
+ if isinstance(self.template["placement"]["managed_cluster"]["config"], str):
+ if urlparse(self.template["placement"]["managed_cluster"]["config"]).scheme == "gs":
+ config = self._build_cluster_config(
+ self.template["placement"]["managed_cluster"]["config"]
+ )
+
+ # Dataproc workflows do not accept life cycle_config
+ if "lifecycle_config" in config:
+ self.log.warning(
+ "Workflow does not accept lifecycle in config, removing it from config"
+ )
+ config.pop("lifecycle_config", None)
+ self.template["placement"]["managed_cluster"]["config"] = config
+ else:
+ self.log.warning(
+ "Workflow Template config accepts only gcs path string or dictioanry type, "
+ "skipping build cluster config from yml file"
+ )
+ except KeyError:
+ pass
Review comment:
Is there a chance that building template will fail on `KeyError` but the template will be valid? If yes, then we should raise the error and expect users to always provided valid templates
--
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] github-actions[bot] closed pull request #18197: Add gcs yml config to create dataproc cluster
Posted by GitBox <gi...@apache.org>.
github-actions[bot] closed pull request #18197:
URL: https://github.com/apache/airflow/pull/18197
--
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] iostreamdoth commented on a change in pull request #18197: Add gcs yml config to create dataproc cluster
Posted by GitBox <gi...@apache.org>.
iostreamdoth commented on a change in pull request #18197:
URL: https://github.com/apache/airflow/pull/18197#discussion_r707831115
##########
File path: airflow/providers/google/cloud/operators/dataproc.py
##########
@@ -48,6 +54,83 @@
)
+class DataProcCreateWorkflowBaseOperator(BaseOperator):
+ """Helper class for preparing configs from workflow template"""
+
+ def __init__(self, **kwargs):
+ super().__init__(**kwargs)
+
+ def prepare_template(self) -> None:
+ try:
+ # if no file is specified, skip
+ if isinstance(self.template["placement"]["managed_cluster"]["config"], str):
+ if urlparse(self.template["placement"]["managed_cluster"]["config"]).scheme == "gs":
+ config = self._build_cluster_config(
+ self.template["placement"]["managed_cluster"]["config"]
+ )
+
+ # Dataproc workflows do not accept life cycle_config
+ if "lifecycle_config" in config:
+ self.log.warning(
+ "Workflow does not accept lifecycle in config, removing it from config"
+ )
+ config.pop("lifecycle_config", None)
+ self.template["placement"]["managed_cluster"]["config"] = config
+ else:
+ self.log.warning(
+ "Workflow Template config accepts only gcs path string or dictioanry type, "
+ "skipping build cluster config from yml file"
+ )
+ except KeyError:
+ pass
Review comment:
A template dictionary can be of 2 types:
1. managed_cluster which needs config key like
```
WORKFLOW_TEMPLATE_YML = {
"id": WORKFLOW_NAME_YML,
"placement": {
"**managed_cluster**": {
"cluster_name": CLUSTER_NAME,
"config": YML_CONFIG_URI,
}
},
"jobs": [{"step_id": "pig_job_1", "pig_job": PIG_JOB["pig_job"]}],
}
```
2. cluster_selector like
```
WORKFLOW_TEMPLATE_YML = {
"id": WORKFLOW_NAME_YML,
"placement": {
"**cluster_selector** ": {
"cluster_name": CLUSTER_NAME
}
},
"jobs": [{"step_id": "pig_job_1", "pig_job": PIG_JOB["pig_job"]}],
}
```
so given that, yes key `self.template["placement"]["**managed_cluster**"]["config"]` can fail in case of `**cluster_selector**` type template but it is still a valid template. I did more digging around it and I feel that prepare_template() is not the right place to perform this conversion of yml config file to dictionary. Will try to resolve the config during execute method.
--
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 #18197: Add gcs yml config to create dataproc cluster
Posted by GitBox <gi...@apache.org>.
SamWheating commented on a change in pull request #18197:
URL: https://github.com/apache/airflow/pull/18197#discussion_r711838154
##########
File path: tests/always/test_project_structure.py
##########
@@ -189,6 +189,7 @@ class TestGoogleProviderProjectStructure(unittest.TestCase):
'airflow.providers.google.cloud.operators.compute.ComputeEngineBaseOperator',
'airflow.providers.google.cloud.operators.cloud_sql.CloudSQLBaseOperator',
'airflow.providers.google.cloud.operators.dataproc.DataprocJobBaseOperator',
+ 'airflow.providers.google.cloud.operators.dataproc.DataProcCreateWorkflowBaseOperator',
Review comment:
```suggestion
'airflow.providers.google.cloud.operators.dataproc.DataprocCreateWorkflowBaseOperator',
```
--
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 #18197: Add gcs yml config to create dataproc cluster
Posted by GitBox <gi...@apache.org>.
SamWheating commented on a change in pull request #18197:
URL: https://github.com/apache/airflow/pull/18197#discussion_r711840591
##########
File path: airflow/providers/google/cloud/operators/dataproc.py
##########
@@ -663,6 +765,16 @@ def execute(self, context) -> dict:
self.log.info('Creating cluster: %s', self.cluster_name)
hook = DataprocHook(gcp_conn_id=self.gcp_conn_id, impersonation_chain=self.impersonation_chain)
# Save data required to display extra link no matter what the cluster status will be
+
+ if isinstance(self.cluster_config, str):
+ if urlparse(self.cluster_config).scheme == "gs":
+ self.cluster_config = self._build_cluster_config(self.cluster_config)
+ else:
+ self.log.info(
Review comment:
Is it worth raising an exception here? If the user provides a non-GCS string then the call to `create_cluster` then I think that the call to `create_cluster` is going to fail, right?
In this case it might be preferable to raise here so that the DAG fails parsing rather than fails during execution.
--
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 #18197: Add gcs yml config to create dataproc cluster
Posted by GitBox <gi...@apache.org>.
SamWheating commented on a change in pull request #18197:
URL: https://github.com/apache/airflow/pull/18197#discussion_r711835811
##########
File path: airflow/providers/google/cloud/operators/dataproc.py
##########
@@ -48,6 +54,101 @@
)
+class DataProcCreateWorkflowBaseOperator(BaseOperator):
Review comment:
```suggestion
class DataprocCreateWorkflowBaseOperator(BaseOperator):
```
For consistency with other dataproc-related classes classes
--
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] iostreamdoth commented on a change in pull request #18197: Add gcs yml config to create dataproc cluster
Posted by GitBox <gi...@apache.org>.
iostreamdoth commented on a change in pull request #18197:
URL: https://github.com/apache/airflow/pull/18197#discussion_r707831115
##########
File path: airflow/providers/google/cloud/operators/dataproc.py
##########
@@ -48,6 +54,83 @@
)
+class DataProcCreateWorkflowBaseOperator(BaseOperator):
+ """Helper class for preparing configs from workflow template"""
+
+ def __init__(self, **kwargs):
+ super().__init__(**kwargs)
+
+ def prepare_template(self) -> None:
+ try:
+ # if no file is specified, skip
+ if isinstance(self.template["placement"]["managed_cluster"]["config"], str):
+ if urlparse(self.template["placement"]["managed_cluster"]["config"]).scheme == "gs":
+ config = self._build_cluster_config(
+ self.template["placement"]["managed_cluster"]["config"]
+ )
+
+ # Dataproc workflows do not accept life cycle_config
+ if "lifecycle_config" in config:
+ self.log.warning(
+ "Workflow does not accept lifecycle in config, removing it from config"
+ )
+ config.pop("lifecycle_config", None)
+ self.template["placement"]["managed_cluster"]["config"] = config
+ else:
+ self.log.warning(
+ "Workflow Template config accepts only gcs path string or dictioanry type, "
+ "skipping build cluster config from yml file"
+ )
+ except KeyError:
+ pass
Review comment:
A template dictionary can be of 2 types:
1. managed_cluster which needs config key like
```
WORKFLOW_TEMPLATE_YML = {
"id": WORKFLOW_NAME_YML,
"placement": {
"**managed_cluster**": {
"cluster_name": CLUSTER_NAME,
"config": YML_CONFIG_URI,
}
},
"jobs": [{"step_id": "pig_job_1", "pig_job": PIG_JOB["pig_job"]}],
}
```
2. cluster_selector like
```
WORKFLOW_TEMPLATE_YML = {
"id": WORKFLOW_NAME_YML,
"placement": {
"**cluster_selector** ": {
"cluster_name": CLUSTER_NAME
}
},
"jobs": [{"step_id": "pig_job_1", "pig_job": PIG_JOB["pig_job"]}],
}
```
so given that, yes key `self.template["placement"]["managed_cluster"]["config"]` can fail in case of `cluster_selector` type template but it is still a valid template. I did more digging around it and I feel that prepare_template() is not the right place to perform this conversion of yml config file to dictionary. Will try to resolve the config during execute method.
--
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] iostreamdoth commented on a change in pull request #18197: Add gcs yml config to create dataproc cluster
Posted by GitBox <gi...@apache.org>.
iostreamdoth commented on a change in pull request #18197:
URL: https://github.com/apache/airflow/pull/18197#discussion_r707831115
##########
File path: airflow/providers/google/cloud/operators/dataproc.py
##########
@@ -48,6 +54,83 @@
)
+class DataProcCreateWorkflowBaseOperator(BaseOperator):
+ """Helper class for preparing configs from workflow template"""
+
+ def __init__(self, **kwargs):
+ super().__init__(**kwargs)
+
+ def prepare_template(self) -> None:
+ try:
+ # if no file is specified, skip
+ if isinstance(self.template["placement"]["managed_cluster"]["config"], str):
+ if urlparse(self.template["placement"]["managed_cluster"]["config"]).scheme == "gs":
+ config = self._build_cluster_config(
+ self.template["placement"]["managed_cluster"]["config"]
+ )
+
+ # Dataproc workflows do not accept life cycle_config
+ if "lifecycle_config" in config:
+ self.log.warning(
+ "Workflow does not accept lifecycle in config, removing it from config"
+ )
+ config.pop("lifecycle_config", None)
+ self.template["placement"]["managed_cluster"]["config"] = config
+ else:
+ self.log.warning(
+ "Workflow Template config accepts only gcs path string or dictioanry type, "
+ "skipping build cluster config from yml file"
+ )
+ except KeyError:
+ pass
Review comment:
A template dictionary can be of 2 types:
1. managed_cluster which needs config key like
```
WORKFLOW_TEMPLATE_YML = {
"id": WORKFLOW_NAME_YML,
"placement": {
"managed_cluster": {
"cluster_name": CLUSTER_NAME,
"config": YML_CONFIG_URI,
}
},
"jobs": [{"step_id": "pig_job_1", "pig_job": PIG_JOB["pig_job"]}],
}
```
2. cluster_selector like which does not need config key
```
WORKFLOW_TEMPLATE_YML = {
"id": WORKFLOW_NAME_YML,
"placement": {
"cluster_selector ": {
"cluster_name": CLUSTER_NAME
}
},
"jobs": [{"step_id": "pig_job_1", "pig_job": PIG_JOB["pig_job"]}],
}
```
so given that, yes key `self.template["placement"]["managed_cluster"]["config"]` can fail in case of `cluster_selector` type template but it is still a valid template. I did more digging around it and I feel that prepare_template() is not the right place to perform this conversion of yml config file to dictionary. Will try to resolve the config during execute method.
--
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 #18197: Add gcs yml config to create dataproc cluster
Posted by GitBox <gi...@apache.org>.
SamWheating commented on a change in pull request #18197:
URL: https://github.com/apache/airflow/pull/18197#discussion_r711837965
##########
File path: airflow/providers/google/cloud/operators/dataproc.py
##########
@@ -663,6 +765,16 @@ def execute(self, context) -> dict:
self.log.info('Creating cluster: %s', self.cluster_name)
hook = DataprocHook(gcp_conn_id=self.gcp_conn_id, impersonation_chain=self.impersonation_chain)
# Save data required to display extra link no matter what the cluster status will be
Review comment:
Could you move this comment below the following block in order to keep it next to the `self.xcom_push()` which it refers 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