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 2022/08/04 18:43:53 UTC
[GitHub] [airflow] blcksrx opened a new pull request, #25541: copy into snowflake from external source
blcksrx opened a new pull request, #25541:
URL: https://github.com/apache/airflow/pull/25541
<!--
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 an existing issue, reference it using one of the following:
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
---
closes: #25395
--
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] blcksrx commented on pull request #25541: copy into snowflake from external source
Posted by GitBox <gi...@apache.org>.
blcksrx commented on PR #25541:
URL: https://github.com/apache/airflow/pull/25541#issuecomment-1206620918
It's true that Airflow connection is different than Snowflake connection but it's possible to create Snowflake connection from Airflow.
In that case I'm going to work on `ExternalStage` operator cause `ExternalLocation` would have leakage in the logs.
--
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] Taragolis commented on pull request #25541: copy into snowflake from external source
Posted by GitBox <gi...@apache.org>.
Taragolis commented on PR #25541:
URL: https://github.com/apache/airflow/pull/25541#issuecomment-1205689098
Yeah, I agree that better implement as some abstract class or just raise an `NotImplementedError` on call specific method
```python
class BaseCopyIntoTable(BaseOperator):
def __init__(self, somearg, some_other_arg, **kwargs):
super.__init__(**kwargs)
...
def from_location(self):
raise NotImplementedError()
def execute(self, context: Context) -> None:
snowflake_hook = SnowflakeHook(...)
sql = f"""
COPY INTO blah.blah FROM self.from_location()
FROM {self.external_location}
Other useful argument for COPY INTO <table>
"""
self.log.info('Executing COPY command...')
snowflake_hook.run(sql)
self.log.info("COPY command completed")
```
And this kind of `BaseCopyIntoTable` could use for multiple different situations and operators (not required implement all of them in one PR)
**COPY INTO without transformation**
1. Specified named internal/external stage
2. Stage for the specified table
3. Stage for the current user
4. Specified external location for supported cloud providers by providing storage integration name (provider agnostic way)
5. Specified external location for AWS S3 by providing credentials, not recommended way and also we need to be sure that credentials not licked to logs. Obtain credentials by [AwsHook](https://github.com/apache/airflow/blob/cda40836ca55702c543391367d0829ed231c1b35/airflow/providers/amazon/aws/hooks/base_aws.py#L525-L535)
6. Specified external location for GCS by providing credentials, not recommended way and also we need to be sure that credentials not licked to logs.
7. Specified external location for Azure Containers by providing credentials, not recommended way and also we need to be sure that credentials not licked to logs.
**COPY INTO with transformation**
1. Specified named internal/external stage
2. Stage for the specified table
3. Stage for the current user
Also it would be nice if this abstract operator supports all subset of snowflake hook parameters which implemented in existed S3ToSnowflakeOperator
https://github.com/apache/airflow/blob/cda40836ca55702c543391367d0829ed231c1b35/airflow/providers/snowflake/transfers/s3_to_snowflake.py#L106-L114
--
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 commented on pull request #25541: copy into snowflake from external source
Posted by GitBox <gi...@apache.org>.
potiuk commented on PR #25541:
URL: https://github.com/apache/airflow/pull/25541#issuecomment-1206605071
I disagree. I think it's better to keep it in Snowflake. First of all by definition (this is our rule of deciding where to add such "transfer" operators is that "target" counts. No matter if it is GCS/S3 etc. everything should be in Snowflake provider.
As far as I understand - you cannot even reuse the connnections that you alrready defined in Airflow, because Airflow connections have different format than the one snowflake uses - and you anyhow would have to do some mapping betweeen then ones in Connection and the ones you send to snowflake.
--
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] eladkal commented on a diff in pull request #25541: copy into snowflake from external source
Posted by GitBox <gi...@apache.org>.
eladkal commented on code in PR #25541:
URL: https://github.com/apache/airflow/pull/25541#discussion_r957802975
##########
airflow/providers/snowflake/transfers/s3_to_snowflake.py:
##########
@@ -85,6 +86,15 @@ def __init__(
session_parameters: Optional[dict] = None,
**kwargs,
) -> None:
+ warnings.warn(
+ """
+ SnowflakeToSlackOperator is deprecated.
Review Comment:
This is wrong class name.
Raised https://github.com/apache/airflow/pull/26047 to fix 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] potiuk merged pull request #25541: copy into snowflake from external source
Posted by GitBox <gi...@apache.org>.
potiuk merged PR #25541:
URL: https://github.com/apache/airflow/pull/25541
--
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 commented on pull request #25541: copy into snowflake from external source
Posted by GitBox <gi...@apache.org>.
potiuk commented on PR #25541:
URL: https://github.com/apache/airflow/pull/25541#issuecomment-1207386822
@potiuk When I mention that I agreed with abstract class I meant that right now current S3ToSnowflakeOperator basically just a query builder for Snowflake COPY INTO <table> and only one piece of query might be changed depend on location (external stage, internal stage, external location), which could be implemented per appropriate class, same as it done in [BaseSQLToGCSOperator](https://github.com/apache/airflow/blob/ae7bf474109410fa838ab2728ae6d581cdd41808/airflow/providers/google/cloud/transfers/sql_to_gcs.py#L316-L326). I absolutely agree with the fact that it should be part of Snowflake Provider.
> That's cool. If we keep'em all in Snowflake, then that is perfectly ok.
And last but not least I do not sure that this operator (current and in this PR) should be a placed in transfers it only generates SQL Statement and everything transfers between Cloud (as well as local stages) complete internally in Snowflake
> Crossed my mind too. I think it shoudl indeed be in the regular `operators` package.
--
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] Taragolis commented on pull request #25541: copy into snowflake from external source
Posted by GitBox <gi...@apache.org>.
Taragolis commented on PR #25541:
URL: https://github.com/apache/airflow/pull/25541#issuecomment-1207385297
@blcksrx Personal thoughts just copy code from [S3ToSnowflakeOperator](https://github.com/apache/airflow/blob/main/airflow/providers/snowflake/transfers/s3_to_snowflake.py#L26-L28) create new operator based, and inherit S3ToSnowflakeOperator by new operator (with deprecation warnings)
Everything except copyOptions and VALIDATION_MODE already implemented there. And it already works with all Cloud Providers which already has configured `externalStage` for appropriate cloud provider (current and futures)
@potiuk When I mention that I agreed with abstract class I meant that right now current `S3ToSnowflakeOperator` basically just a query builder for Snowflake `COPY INTO <table>` and only one piece of query might be changed depend on location (external stage, internal stage, external location), which could be implemented per appropriate class, same as it done in [BaseSQLToGCSOperator](https://github.com/apache/airflow/blob/ae7bf474109410fa838ab2728ae6d581cdd41808/airflow/providers/google/cloud/transfers/sql_to_gcs.py#L316-L326). I absolutely agree with the fact that it should be part of Snowflake Provider.
And last but not least I do not sure that this operator (current and in this PR) should be a placed in `transfers` it only generates SQL Statement and everything transfers between Cloud (as well as local stages) complete internally in Snowflake
--
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 commented on pull request #25541: copy into snowflake from external source
Posted by GitBox <gi...@apache.org>.
potiuk commented on PR #25541:
URL: https://github.com/apache/airflow/pull/25541#issuecomment-1207410528
> But I don't think this a good idea to implement operator which pass credentials directly to SQL query (not as part Snowflake providers and any others). This credentials could expose to logs and also it not recommended by Snowflake docs:
Absolutely.
> However if we implement COPY INTO <table> as some kind of abstract operator than end users might create their own custom operator to fit their requirements and provide credentials as part of SQL query if they do not won't follow recommendation and security practice.
Yes but we should discourage that in our docs as well, pointing to SF docs.
--
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] Taragolis commented on pull request #25541: copy into snowflake from external source
Posted by GitBox <gi...@apache.org>.
Taragolis commented on PR #25541:
URL: https://github.com/apache/airflow/pull/25541#issuecomment-1207392425
> Crossed my mind too. I think it shoudl indeed be in the regular operators package.
The only one case when it could be as `transfers` when query use credentials (instead of stages or integrations) for AWS or Azure (GCS not supported), like
```sql
COPY INTO FOO.BAR
FROM s3://source-bucket-123/key/path/location
CREDENTIALS = ( AWS_KEY_ID='xxxx' AWS_SECRET_KEY='xxxxx' AWS_TOKEN='xxxxxx' )
FILE_FORMAT = ( TYPE = PARQUET )
;
```
```sql
COPY INTO FOO.BAR
FROM azure://{account}.blob.core.windows.net/source-container-123/key/path/location
CREDENTIALS = ( AZURE_SAS_TOKEN = 'xxx' )
FILE_FORMAT = ( TYPE = PARQUET )
;
```
But **I don't think this a good idea** to implement operator which pass credentials directly to SQL query (not as part Snowflake providers and any others). This credentials could expose to logs and also it not recommended by Snowflake docs:
```
We highly recommend the use of storage integrations.
This option avoids the need to supply cloud storage credentials using the CREDENTIALS parameter when creating stages or loading data.
```
However if we implement `COPY INTO <table>` as some kind of abstract operator than end users might create their own custom operator to fit their requirements and provide credentials as part of SQL query if they do not won't follow recommendation and security practice.
--
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] blcksrx commented on pull request #25541: copy into snowflake from external source
Posted by GitBox <gi...@apache.org>.
blcksrx commented on PR #25541:
URL: https://github.com/apache/airflow/pull/25541#issuecomment-1205650422
@potiuk
In my opinion, it's better to make this operator as an abstract class and implement it separately [just overriding external location section] on each cloud provider cause I believe from user experience it's not pleasant if he/she already created a cloud connection such as AWS and does not use it on this operator beside of the security breach on the sql part that needs to pass the credentials as args.
In addition, using `externalStage` needs manual effort on the user side to first create a stage then use it on the airflow that reduces the flexibility.
P.S: I created this draft PR to let you know this issue and know your opinion about it
c.c: @Taragolis
--
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] Taragolis commented on a diff in pull request #25541: copy into snowflake from external source
Posted by GitBox <gi...@apache.org>.
Taragolis commented on code in PR #25541:
URL: https://github.com/apache/airflow/pull/25541#discussion_r939644872
##########
airflow/providers/snowflake/transfers/s3_to_snowflake.py:
##########
@@ -19,11 +19,17 @@
"""This module contains AWS S3 to Snowflake operator."""
from typing import Any, Optional, Sequence
+from deprecated import deprecated
+
from airflow.models import BaseOperator
from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook
from airflow.providers.snowflake.utils.common import enclose_param
+@deprecated(
+ reason="use airflow.providers.snowflake.transfers.copy_into_snowflake.CopyFromExternalStageToSnowflake",
+ version="2.3.4",
+)
Review Comment:
It won't work, this module not as part of Airflow dependencies.
And also providers has different lifecycle: releases and deprecations which most independent of core Airflow - should support minimum Airflow 2.2 (in general).
So better use `watnings.warn` for depreciation inside of `__init__` method, example:
https://github.com/apache/airflow/blob/1708da9233c13c3821d76e56dbe0e383ff67b0fd/airflow/providers/snowflake/transfers/snowflake_to_slack.py#L87-L94
##########
airflow/providers/snowflake/transfers/copy_into_snowflake.py:
##########
@@ -0,0 +1,149 @@
+#
+# 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 abstract operator that child classes implements
+COPY INTO <TABLE> SQL in Snowflake
+"""
+
+
+from typing import List, Optional, Sequence
+
+from airflow.models import BaseOperator
+from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook
+from airflow.providers.snowflake.utils.common import enclose_param
+from airflow.utils.context import Context
+
+
+class CopyFromExternalStageToSnowflakeOperator(BaseOperator):
+ """
+ Executes an COPY INTO command to load files from an external stage from clouds to Snowflake
+ .. seealso::
+ `COPY INTO Syntax: <https://docs.snowflake.com/en/sql-reference/sql/copy-into-table.html>`_
+ `CREATE STAGE Syntax: <https://docs.snowflake.com/en/sql-reference/sql/create-stage.html>`_
+
+ This operator requires the snowflake_conn_id connection. The snowflake host, login,
+ and, password field must be setup in the connection. Other inputs can be defined
+ in the connection or hook instantiation.
+
+ :param namespace: snowflake namespace
+ :param table: snowflake table
+ :param file_format: file format name i.e. CSV, AVRO, etc
+ :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
+ :param prefix: cloud storage location specified to limit the set of files to load
+ :param files: files to load into table
+ :param pattern: pattern to load files from external location to table
+ :param copy_into_postifx: optional sql postfix for INSERT INTO query
+ such as formatTypeOptions and copyOptions
+ :param snowflake_conn_id: Reference to :ref:`Snowflake connection id<howto/connection:snowflake>`
+ :param account: snowflake account name
+ :param warehouse: name of snowflake warehouse
+ :param database: name of snowflake database
+ :param region: name of snowflake region
+ :param role: name of snowflake role
+ :param schema: name of snowflake schema
+ :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.
+ :param session_parameters: You can set session-level parameters at
+ the time you connect to Snowflake
+
+ """
+
+ template_fields: Sequence[str] = ("files",)
+ template_fields_renderers = {"files": "json"}
+
+ def __init__(
+ self,
+ table: str,
+ file_format: str,
+ stage: str,
+ stage_postfix: Optional[str] = None,
+ columns_array: Optional[List[str]] = None,
+ files: Optional[List[str]] = None,
+ pattern: Optional[str] = None,
+ copy_into_postfix: Optional[str] = None,
+ snowflake_conn_id: str = "snowflake_default",
+ account: Optional[str] = None,
+ warehouse: Optional[str] = None,
+ database: Optional[str] = None,
+ region: Optional[str] = None,
+ role: Optional[str] = None,
+ schema: Optional[str] = None,
+ authenticator: Optional[str] = None,
+ session_parameters: Optional[dict] = None,
+ autocommit: bool = True,
+ *args,
+ **kwargs,
+ ):
+ super().__init__(*args, **kwargs)
+ self.table = table
+ self.file_format = file_format
+ self.stage = stage
+ self.stage_postfix = stage_postfix
+ self.columns_array = columns_array
+ self.files = files
+ self.pattern = pattern
+ self.copy_into_postfix = copy_into_postfix
+ self.snowflake_conn_id = snowflake_conn_id
+ self.account = account
+ self.warehouse = warehouse
+ self.database = database
+ self.region = region
+ self.role = role
+ self.schema = schema
+ self.authenticator = authenticator
+ self.session_parameters = session_parameters
+ self.autocommit = autocommit
+
+ def execute(self, context: Context) -> None:
+ snowflake_hook = SnowflakeHook(
+ snowflake_conn_id=self.snowflake_conn_id,
+ account=self.account,
+ warehouse=self.warehouse,
+ database=self.database,
+ region=self.region,
+ role=self.role,
+ schema=self.schema,
+ authenticator=self.authenticator,
+ session_parameters=self.session_parameters,
+ )
+
+ if self.schema:
+ into = f"{self.schema}.{self.table}"
+ else:
+ into = self.table
+
+ if self.columns_array:
+ into = f"{into}({', '.join(self.columns_array)})"
+
+ sql = f"""
+ COPY INTO {into}
+ FROM @{self.stage}/{self.stage_postfix or ""}
+ {"FILES=" + ",".join(map(enclose_param ,self.files)) if self.files else ""}
+ {"PATTERN=" + enclose_param(self.pattern) if self.pattern else ""}
+ FILE_FORMAT={self.file_format}
+ {self.copy_into_postfix or ""}
+ """
Review Comment:
Is it really required to use a multi-line string syntax rather than prepare as string in parts, as it was done in `S3ToSnowflakeOperator` ?
##########
airflow/providers/snowflake/transfers/copy_into_snowflake.py:
##########
@@ -0,0 +1,146 @@
+#
+# 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 abstract operator that child classes implements
+COPY INTO <TABLE> SQL in Snowflake
+"""
+
+
+from typing import List, Optional, Sequence
+
+from airflow.models import BaseOperator
+from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook
+from airflow.providers.snowflake.utils.common import enclose_param
+from airflow.utils.context import Context
+
+
+class CopyFromExternalStageToSnowflakeOperator(BaseOperator):
+ """
+ Executes a COPY INTO command to load files from an external stage from clouds to Snowflake
+
+ This operator requires the snowflake_conn_id connection. The snowflake host, login,
+ and, password field must be setup in the connection. Other inputs can be defined
+ in the connection or hook instantiation.
+
+ :param namespace: snowflake namespace
+ :param table: snowflake table
+ :param file_format: file format name i.e. CSV, AVRO, etc
+ :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
+ :param prefix: cloud storage location specified to limit the set of files to load
+ :param files: files to load into table
+ :param pattern: pattern to load files from external location to table
+ :param copy_into_postifx: optional sql postfix for INSERT INTO query
+ such as `formatTypeOptions` and `copyOptions`
+ :param snowflake_conn_id: Reference to :ref:`Snowflake connection id<howto/connection:snowflake>`
+ :param account: snowflake account name
+ :param warehouse: name of snowflake warehouse
+ :param database: name of snowflake database
+ :param region: name of snowflake region
+ :param role: name of snowflake role
+ :param schema: name of snowflake schema
+ :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.
+ :param session_parameters: You can set session-level parameters at
+ the time you connect to Snowflake
+
+ """
+
+ template_fields: Sequence[str] = ("files",)
+ template_fields_renderers = {"files": "json"}
+
+ def __init__(
+ self,
+ table: str,
+ file_format: str,
+ stage: str,
+ stage_postfix: Optional[str] = None,
+ columns_array: Optional[List[str]] = None,
+ files: Optional[List[str]] = None,
+ pattern: Optional[str] = None,
+ copy_into_postfix: Optional[str] = None,
Review Comment:
Might be better use `copy_options` and `validation_mode` in order to conform with [Snowflake COPY INTO syntax](https://docs.snowflake.com/en/sql-reference/sql/copy-into-table.html#syntax)
##########
airflow/providers/snowflake/transfers/copy_into_snowflake.py:
##########
@@ -0,0 +1,149 @@
+#
+# 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 abstract operator that child classes implements
+COPY INTO <TABLE> SQL in Snowflake
+"""
+
+
+from typing import List, Optional, Sequence
+
+from airflow.models import BaseOperator
+from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook
+from airflow.providers.snowflake.utils.common import enclose_param
+from airflow.utils.context import Context
+
+
+class CopyFromExternalStageToSnowflakeOperator(BaseOperator):
+ """
+ Executes an COPY INTO command to load files from an external stage from clouds to Snowflake
+ .. seealso::
+ `COPY INTO Syntax: <https://docs.snowflake.com/en/sql-reference/sql/copy-into-table.html>`_
+ `CREATE STAGE Syntax: <https://docs.snowflake.com/en/sql-reference/sql/create-stage.html>`_
+
+ This operator requires the snowflake_conn_id connection. The snowflake host, login,
+ and, password field must be setup in the connection. Other inputs can be defined
+ in the connection or hook instantiation.
+
+ :param namespace: snowflake namespace
+ :param table: snowflake table
+ :param file_format: file format name i.e. CSV, AVRO, etc
+ :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
+ :param prefix: cloud storage location specified to limit the set of files to load
+ :param files: files to load into table
+ :param pattern: pattern to load files from external location to table
+ :param copy_into_postifx: optional sql postfix for INSERT INTO query
+ such as formatTypeOptions and copyOptions
+ :param snowflake_conn_id: Reference to :ref:`Snowflake connection id<howto/connection:snowflake>`
+ :param account: snowflake account name
+ :param warehouse: name of snowflake warehouse
+ :param database: name of snowflake database
+ :param region: name of snowflake region
+ :param role: name of snowflake role
+ :param schema: name of snowflake schema
+ :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.
+ :param session_parameters: You can set session-level parameters at
+ the time you connect to Snowflake
+
+ """
+
+ template_fields: Sequence[str] = ("files",)
+ template_fields_renderers = {"files": "json"}
+
+ def __init__(
+ self,
+ table: str,
+ file_format: str,
+ stage: str,
+ stage_postfix: Optional[str] = None,
Review Comment:
I better keep the previous name (actually right now it uses in docstring) `prefix`
--
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] Taragolis commented on a diff in pull request #25541: copy into snowflake from external source
Posted by GitBox <gi...@apache.org>.
Taragolis commented on code in PR #25541:
URL: https://github.com/apache/airflow/pull/25541#discussion_r939647407
##########
airflow/providers/snowflake/transfers/copy_into_snowflake.py:
##########
@@ -0,0 +1,146 @@
+#
+# 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 abstract operator that child classes implements
+COPY INTO <TABLE> SQL in Snowflake
+"""
+
+
+from typing import List, Optional, Sequence
+
+from airflow.models import BaseOperator
+from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook
+from airflow.providers.snowflake.utils.common import enclose_param
+from airflow.utils.context import Context
+
+
+class CopyFromExternalStageToSnowflakeOperator(BaseOperator):
+ """
+ Executes a COPY INTO command to load files from an external stage from clouds to Snowflake
+
+ This operator requires the snowflake_conn_id connection. The snowflake host, login,
+ and, password field must be setup in the connection. Other inputs can be defined
+ in the connection or hook instantiation.
+
+ :param namespace: snowflake namespace
+ :param table: snowflake table
+ :param file_format: file format name i.e. CSV, AVRO, etc
+ :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
+ :param prefix: cloud storage location specified to limit the set of files to load
+ :param files: files to load into table
+ :param pattern: pattern to load files from external location to table
+ :param copy_into_postifx: optional sql postfix for INSERT INTO query
+ such as `formatTypeOptions` and `copyOptions`
+ :param snowflake_conn_id: Reference to :ref:`Snowflake connection id<howto/connection:snowflake>`
+ :param account: snowflake account name
+ :param warehouse: name of snowflake warehouse
+ :param database: name of snowflake database
+ :param region: name of snowflake region
+ :param role: name of snowflake role
+ :param schema: name of snowflake schema
+ :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.
+ :param session_parameters: You can set session-level parameters at
+ the time you connect to Snowflake
+
+ """
+
+ template_fields: Sequence[str] = ("files",)
+ template_fields_renderers = {"files": "json"}
+
+ def __init__(
+ self,
+ table: str,
+ file_format: str,
+ stage: str,
+ stage_postfix: Optional[str] = None,
+ columns_array: Optional[List[str]] = None,
+ files: Optional[List[str]] = None,
+ pattern: Optional[str] = None,
+ copy_into_postfix: Optional[str] = None,
Review Comment:
Might be better use `copy_options` and `validation_mode` rather than `copy_into_postfix`, in order to conform with [Snowflake COPY INTO syntax](https://docs.snowflake.com/en/sql-reference/sql/copy-into-table.html#syntax)
--
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] blcksrx commented on a diff in pull request #25541: copy into snowflake from external source
Posted by GitBox <gi...@apache.org>.
blcksrx commented on code in PR #25541:
URL: https://github.com/apache/airflow/pull/25541#discussion_r944344190
##########
airflow/providers/snowflake/transfers/copy_into_snowflake.py:
##########
@@ -0,0 +1,149 @@
+#
+# 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 abstract operator that child classes implements
+COPY INTO <TABLE> SQL in Snowflake
+"""
+
+
+from typing import List, Optional, Sequence
+
+from airflow.models import BaseOperator
+from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook
+from airflow.providers.snowflake.utils.common import enclose_param
+from airflow.utils.context import Context
+
+
+class CopyFromExternalStageToSnowflakeOperator(BaseOperator):
+ """
+ Executes an COPY INTO command to load files from an external stage from clouds to Snowflake
+ .. seealso::
+ `COPY INTO Syntax: <https://docs.snowflake.com/en/sql-reference/sql/copy-into-table.html>`_
+ `CREATE STAGE Syntax: <https://docs.snowflake.com/en/sql-reference/sql/create-stage.html>`_
+
+ This operator requires the snowflake_conn_id connection. The snowflake host, login,
+ and, password field must be setup in the connection. Other inputs can be defined
+ in the connection or hook instantiation.
+
+ :param namespace: snowflake namespace
+ :param table: snowflake table
+ :param file_format: file format name i.e. CSV, AVRO, etc
+ :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
+ :param prefix: cloud storage location specified to limit the set of files to load
+ :param files: files to load into table
+ :param pattern: pattern to load files from external location to table
+ :param copy_into_postifx: optional sql postfix for INSERT INTO query
+ such as formatTypeOptions and copyOptions
+ :param snowflake_conn_id: Reference to :ref:`Snowflake connection id<howto/connection:snowflake>`
+ :param account: snowflake account name
+ :param warehouse: name of snowflake warehouse
+ :param database: name of snowflake database
+ :param region: name of snowflake region
+ :param role: name of snowflake role
+ :param schema: name of snowflake schema
+ :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.
+ :param session_parameters: You can set session-level parameters at
+ the time you connect to Snowflake
+
+ """
+
+ template_fields: Sequence[str] = ("files",)
+ template_fields_renderers = {"files": "json"}
+
+ def __init__(
+ self,
+ table: str,
+ file_format: str,
+ stage: str,
+ stage_postfix: Optional[str] = None,
+ columns_array: Optional[List[str]] = None,
+ files: Optional[List[str]] = None,
+ pattern: Optional[str] = None,
+ copy_into_postfix: Optional[str] = None,
+ snowflake_conn_id: str = "snowflake_default",
+ account: Optional[str] = None,
+ warehouse: Optional[str] = None,
+ database: Optional[str] = None,
+ region: Optional[str] = None,
+ role: Optional[str] = None,
+ schema: Optional[str] = None,
+ authenticator: Optional[str] = None,
+ session_parameters: Optional[dict] = None,
+ autocommit: bool = True,
+ *args,
+ **kwargs,
+ ):
+ super().__init__(*args, **kwargs)
+ self.table = table
+ self.file_format = file_format
+ self.stage = stage
+ self.stage_postfix = stage_postfix
+ self.columns_array = columns_array
+ self.files = files
+ self.pattern = pattern
+ self.copy_into_postfix = copy_into_postfix
+ self.snowflake_conn_id = snowflake_conn_id
+ self.account = account
+ self.warehouse = warehouse
+ self.database = database
+ self.region = region
+ self.role = role
+ self.schema = schema
+ self.authenticator = authenticator
+ self.session_parameters = session_parameters
+ self.autocommit = autocommit
+
+ def execute(self, context: Context) -> None:
+ snowflake_hook = SnowflakeHook(
+ snowflake_conn_id=self.snowflake_conn_id,
+ account=self.account,
+ warehouse=self.warehouse,
+ database=self.database,
+ region=self.region,
+ role=self.role,
+ schema=self.schema,
+ authenticator=self.authenticator,
+ session_parameters=self.session_parameters,
+ )
+
+ if self.schema:
+ into = f"{self.schema}.{self.table}"
+ else:
+ into = self.table
+
+ if self.columns_array:
+ into = f"{into}({', '.join(self.columns_array)})"
+
+ sql = f"""
+ COPY INTO {into}
+ FROM @{self.stage}/{self.stage_postfix or ""}
+ {"FILES=" + ",".join(map(enclose_param ,self.files)) if self.files else ""}
+ {"PATTERN=" + enclose_param(self.pattern) if self.pattern else ""}
+ FILE_FORMAT={self.file_format}
+ {self.copy_into_postfix or ""}
+ """
Review Comment:
I rather to use multiline f-string, it's more clear of string block construction
--
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