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/12/22 10:29:29 UTC
[GitHub] [airflow] aa3pankaj opened a new pull request #20459: Adding Snowflake to S3 transfer operator, Updates in S3 to snowflake
aa3pankaj opened a new pull request #20459:
URL: https://github.com/apache/airflow/pull/20459
<!--
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/
-->
- Adding new transfer operator for Snowflake to S3
- Adding on_error option in S3 to snowflake
- Updates related to query_ids and execution_info
---
**^ 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] aa3pankaj commented on pull request #20459: Adding snowflake_to_s3 transfer operator, Updates in s3_to_snowflake
Posted by GitBox <gi...@apache.org>.
aa3pankaj commented on pull request #20459:
URL: https://github.com/apache/airflow/pull/20459#issuecomment-1000482749
@mik-laj @potiuk @turbaszek any update on this PR?
This consists of changes in already merged operator S3ToSnowflake also, could you guys please review the changes?
--
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] aa3pankaj commented on a change in pull request #20459: Adding snowflake_to_s3 transfer operator, Updates in s3_to_snowflake
Posted by GitBox <gi...@apache.org>.
aa3pankaj commented on a change in pull request #20459:
URL: https://github.com/apache/airflow/pull/20459#discussion_r775921457
##########
File path: airflow/providers/snowflake/transfers/snowflake_to_s3.py
##########
@@ -0,0 +1,153 @@
+#
+# 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 Snowflake to AWS S3 operator."""
+from typing import Any, List, Optional
+
+from airflow.models import BaseOperator
+from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook
+
+
+class SnowflakeToS3Operator(BaseOperator):
+ """
+ Executes COPY command to unload data from Snowflake to s3
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the guide:
+ :ref:`howto/operator:SnowflakeToS3Operator`
+
+ :param stage: reference to a specific snowflake stage. If the stage's schema is not the same as the
+ table one, it must be specified
+ :type stage: str
+ :param prefix: cloud storage location specified to limit the set of files to load
+ :type prefix: str
+ :param file_format: reference to a specific file format
+ :type file_format: str
+ :param on_error: action to be taken in case of any error, possible options are { CONTINUE | SKIP_FILE |
+ SKIP_FILE_<num> | SKIP_FILE_<num>% | ABORT_STATEMENT }
+ :type on_error: str
+ :param header: whether header (column names) needed or not in the unloaded data
+ :type header: bool
+ :param overwrite: whether unloaded data should be overwritten or not in case of same prefix
+ :type overwrite: bool
+ :param single: whether unloaded data should be in a single file or multiple files,
+ default snowflake behaviour is multiple
+ :type single: bool
+ :param unload_sql: sql that would be used for unloading data
+ :type unload_sql: str
+ :param warehouse: name of warehouse (will overwrite any warehouse
+ defined in the connection's extra JSON)
+ :type warehouse: str
+ :param database: reference to a specific database in Snowflake connection
+ :type database: str
+ :param snowflake_conn_id: Reference to
+ :ref:`Snowflake connection id<howto/connection:snowflake>`
+ :type snowflake_conn_id: str
+ :param role: name of role (will overwrite any role defined in
+ connection's extra JSON)
+ :type role: str
+ :param authenticator: authenticator for Snowflake.
+ 'snowflake' (default) to use the internal Snowflake authenticator
+ 'externalbrowser' to authenticate using your web browser and
+ Okta, ADFS or any other SAML 2.0-compliant identify provider
+ (IdP) that has been defined for your account
+ 'https://<your_okta_account_name>.okta.com' to authenticate
+ through native Okta.
+ :type authenticator: str
+ :param session_parameters: You can set session-level parameters at
+ the time you connect to Snowflake
+ :type session_parameters: dict
+ """
+
+ def __init__(
+ self,
+ *,
+ stage: str,
+ prefix: Optional[str] = None,
+ file_format: str,
+ on_error: Optional[str] = None,
+ unload_sql: str,
+ header: Optional[bool] = False,
+ single: Optional[bool] = False,
+ overwrite: Optional[bool] = False,
+ schema: Optional[str] = None,
+ warehouse: Optional[str] = None,
+ database: Optional[str] = None,
+ autocommit: bool = True,
+ snowflake_conn_id: str = 'snowflake_default',
+ role: Optional[str] = None,
+ authenticator: Optional[str] = None,
+ session_parameters: Optional[dict] = None,
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.warehouse = warehouse
+ self.database = database
+ self.stage = stage
+ self.prefix = prefix
+ self.file_format = file_format
+ self.on_error = on_error
+ self.unload_sql = unload_sql
+ self.header = header
+ self.single = single
+ self.overwrite = overwrite
+ self.schema = schema
+ self.autocommit = autocommit
+ self.snowflake_conn_id = snowflake_conn_id
+ self.role = role
+ self.authenticator = authenticator
+ self.session_parameters = session_parameters
+ self.query_ids: List[str] = []
+
+ def execute(self, context: Any) -> None:
+ snowflake_hook = SnowflakeHook(
+ snowflake_conn_id=self.snowflake_conn_id,
+ warehouse=self.warehouse,
+ database=self.database,
+ role=self.role,
+ schema=self.schema,
+ authenticator=self.authenticator,
+ session_parameters=self.session_parameters,
+ )
+
+ sql_parts = [
+ f"COPY INTO @{self.stage}/{self.prefix or ''}",
Review comment:
Here we expect user to pass valid identifier, anyway do you expect SnowflakeToS3Operator to validate stage name using some regex?
--
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] aa3pankaj closed pull request #20459: Adding snowflake_to_s3 transfer operator, Updates in s3_to_snowflake
Posted by GitBox <gi...@apache.org>.
aa3pankaj closed pull request #20459:
URL: https://github.com/apache/airflow/pull/20459
--
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 #20459: Adding snowflake_to_s3 transfer operator, Updates in s3_to_snowflake
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #20459:
URL: https://github.com/apache/airflow/pull/20459#issuecomment-1036815777
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] boring-cyborg[bot] commented on pull request #20459: Adding Snowflake to S3 transfer operator, Updates in S3 to snowflake
Posted by GitBox <gi...@apache.org>.
boring-cyborg[bot] commented on pull request #20459:
URL: https://github.com/apache/airflow/pull/20459#issuecomment-999464261
Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst)
Here are some useful points:
- Pay attention to the quality of your code (flake8, mypy and type annotations). Our [pre-commits]( https://github.com/apache/airflow/blob/main/STATIC_CODE_CHECKS.rst#prerequisites-for-pre-commit-hooks) will help you with that.
- In case of a new feature add useful documentation (in docstrings or in `docs/` directory). Adding a new operator? Check this short [guide](https://github.com/apache/airflow/blob/main/docs/apache-airflow/howto/custom-operator.rst) Consider adding an example DAG that shows how users should use it.
- Consider using [Breeze environment](https://github.com/apache/airflow/blob/main/BREEZE.rst) for testing locally, itโs a heavy docker but it ships with a working Airflow and a lot of integrations.
- Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
- Please follow [ASF Code of Conduct](https://www.apache.org/foundation/policies/conduct) for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
- Be sure to read the [Airflow Coding style]( https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#coding-style-and-best-practices).
Apache Airflow is a community-driven project and together we are making it better ๐.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack
--
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] aa3pankaj commented on pull request #20459: Adding snowflake_to_s3 transfer operator, Updates in s3_to_snowflake
Posted by GitBox <gi...@apache.org>.
aa3pankaj commented on pull request #20459:
URL: https://github.com/apache/airflow/pull/20459#issuecomment-999709969
> Why just SnowflakeOperator is not enough? The operator logic is quite simple and I think we can expect the user to just copy the SQL query from the Snowflake documentation.
@mik-laj We need this for the same reason why we have SnowflakeOperator when we can use SnowflakeHook directly to run queries.
And S3ToSnowflake related changes are required since we already have this operator merged, and currently query_ids and execution_info not exposed from this operator.
--
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] JavierLopezT commented on pull request #20459: Adding snowflake_to_s3 transfer operator, Updates in s3_to_snowflake
Posted by GitBox <gi...@apache.org>.
JavierLopezT commented on pull request #20459:
URL: https://github.com/apache/airflow/pull/20459#issuecomment-1001012984
@aa3pankaj I already tried to merge the SnowflakeToS3Operator and it seems that you would have to create a generic operator SnowflakeToStorageOperator. You can read the comments of my already closed pul request: https://github.com/apache/airflow/pull/14415
--
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 #20459: Adding snowflake_to_s3 transfer operator, Updates in s3_to_snowflake
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #20459:
URL: https://github.com/apache/airflow/pull/20459#discussion_r775208864
##########
File path: airflow/providers/snowflake/transfers/snowflake_to_s3.py
##########
@@ -0,0 +1,153 @@
+#
+# 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 Snowflake to AWS S3 operator."""
+from typing import Any, List, Optional
+
+from airflow.models import BaseOperator
+from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook
+
+
+class SnowflakeToS3Operator(BaseOperator):
+ """
+ Executes COPY command to unload data from Snowflake to s3
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the guide:
+ :ref:`howto/operator:SnowflakeToS3Operator`
+
+ :param stage: reference to a specific snowflake stage. If the stage's schema is not the same as the
+ table one, it must be specified
+ :type stage: str
+ :param prefix: cloud storage location specified to limit the set of files to load
+ :type prefix: str
+ :param file_format: reference to a specific file format
+ :type file_format: str
+ :param on_error: action to be taken in case of any error, possible options are { CONTINUE | SKIP_FILE |
+ SKIP_FILE_<num> | SKIP_FILE_<num>% | ABORT_STATEMENT }
+ :type on_error: str
+ :param header: whether header (column names) needed or not in the unloaded data
+ :type header: bool
+ :param overwrite: whether unloaded data should be overwritten or not in case of same prefix
+ :type overwrite: bool
+ :param single: whether unloaded data should be in a single file or multiple files,
+ default snowflake behaviour is multiple
+ :type single: bool
+ :param unload_sql: sql that would be used for unloading data
+ :type unload_sql: str
+ :param warehouse: name of warehouse (will overwrite any warehouse
+ defined in the connection's extra JSON)
+ :type warehouse: str
+ :param database: reference to a specific database in Snowflake connection
+ :type database: str
+ :param snowflake_conn_id: Reference to
+ :ref:`Snowflake connection id<howto/connection:snowflake>`
+ :type snowflake_conn_id: str
+ :param role: name of role (will overwrite any role defined in
+ connection's extra JSON)
+ :type role: str
+ :param authenticator: authenticator for Snowflake.
+ 'snowflake' (default) to use the internal Snowflake authenticator
+ 'externalbrowser' to authenticate using your web browser and
+ Okta, ADFS or any other SAML 2.0-compliant identify provider
+ (IdP) that has been defined for your account
+ 'https://<your_okta_account_name>.okta.com' to authenticate
+ through native Okta.
+ :type authenticator: str
+ :param session_parameters: You can set session-level parameters at
+ the time you connect to Snowflake
+ :type session_parameters: dict
+ """
+
+ def __init__(
+ self,
+ *,
+ stage: str,
+ prefix: Optional[str] = None,
+ file_format: str,
+ on_error: Optional[str] = None,
+ unload_sql: str,
+ header: Optional[bool] = False,
+ single: Optional[bool] = False,
+ overwrite: Optional[bool] = False,
+ schema: Optional[str] = None,
+ warehouse: Optional[str] = None,
+ database: Optional[str] = None,
+ autocommit: bool = True,
+ snowflake_conn_id: str = 'snowflake_default',
+ role: Optional[str] = None,
+ authenticator: Optional[str] = None,
+ session_parameters: Optional[dict] = None,
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.warehouse = warehouse
+ self.database = database
+ self.stage = stage
+ self.prefix = prefix
+ self.file_format = file_format
+ self.on_error = on_error
+ self.unload_sql = unload_sql
+ self.header = header
+ self.single = single
+ self.overwrite = overwrite
+ self.schema = schema
+ self.autocommit = autocommit
+ self.snowflake_conn_id = snowflake_conn_id
+ self.role = role
+ self.authenticator = authenticator
+ self.session_parameters = session_parameters
+ self.query_ids: List[str] = []
+
+ def execute(self, context: Any) -> None:
+ snowflake_hook = SnowflakeHook(
+ snowflake_conn_id=self.snowflake_conn_id,
+ warehouse=self.warehouse,
+ database=self.database,
+ role=self.role,
+ schema=self.schema,
+ authenticator=self.authenticator,
+ session_parameters=self.session_parameters,
+ )
+
+ sql_parts = [
+ f"COPY INTO @{self.stage}/{self.prefix or ''}",
Review comment:
Stage is the identifier of the object, so it can contain spaces. We should try to ensure that this identifier is safely passed to the query being built, e.g. the use of spaces in the identifier does not cause problems. See: https://docs.snowflake.com/en/sql-reference/identifiers-syntax.html
--
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 #20459: Adding snowflake_to_s3 transfer operator, Updates in s3_to_snowflake
Posted by GitBox <gi...@apache.org>.
mik-laj commented on pull request #20459:
URL: https://github.com/apache/airflow/pull/20459#issuecomment-999684695
Why just SnowflakeOperator is not enough? The operator logic is quite simple and I think we can expect the user to just copy the SQL query from the Snowflake documentation.
--
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 #20459: Adding snowflake_to_s3 transfer operator, Updates in s3_to_snowflake
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #20459:
URL: https://github.com/apache/airflow/pull/20459#discussion_r775209069
##########
File path: tests/providers/snowflake/transfers/test_snowflake_to_s3.py
##########
@@ -0,0 +1,75 @@
+#
+# 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.
+
+from unittest import mock
+
+import pytest
+
+from airflow.providers.snowflake.transfers.snowflake_to_s3 import SnowflakeToS3Operator
+
+
+class TestSnowflakeToS3Transfer:
+ @pytest.mark.parametrize("prefix", [None, 'prefix'])
+ @pytest.mark.parametrize("schema", [None, 'schema'])
+ @pytest.mark.parametrize("unload_sql", ['unload_sql'])
+ @pytest.mark.parametrize("on_error", [None, 'CONTINUE', 'SKIP_FILE', 'ABORT_STATEMENT'])
+ @pytest.mark.parametrize("header", [None, True, False])
+ @pytest.mark.parametrize("overwrite", [None, True, False])
+ @pytest.mark.parametrize("single", [None, True, False])
+ @mock.patch("airflow.providers.snowflake.hooks.snowflake.SnowflakeHook.run")
+ def test_execute(self, mock_run, schema, prefix, unload_sql, on_error, header, overwrite, single):
Review comment:
I don't understand this test. This doesn't test almost anything but is mostly a duplicate of operator logic. We should copy a few generated statements instead of repeating the statement generating code.
--
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] aa3pankaj edited a comment on pull request #20459: Adding snowflake_to_s3 transfer operator, Updates in s3_to_snowflake
Posted by GitBox <gi...@apache.org>.
aa3pankaj edited a comment on pull request #20459:
URL: https://github.com/apache/airflow/pull/20459#issuecomment-999709969
> Why just SnowflakeOperator is not enough? The operator logic is quite simple and I think we can expect the user to just copy the SQL query from the Snowflake documentation.
@mik-laj We need this for the same reason why we have the SnowflakeOperator when we can use SnowflakeHook directly to run queries.
And S3ToSnowflake related changes are required since we already have this operator merged, and currently query_ids and execution_info not exposed from this operator.
--
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