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/02/21 06:09:20 UTC
[GitHub] [airflow] rsg17 opened a new pull request #21704: Add GCSToTrinoOperator
rsg17 opened a new pull request #21704:
URL: https://github.com/apache/airflow/pull/21704
Follow-up PR as discussed in https://github.com/apache/airflow/pull/21084
Logic followed is similar to the above PR.
Loads a csv file from Google Cloud Storage into a Trino table.
Assumptions:
1. First row of the csv contains headers
2. Trino table with requisite columns is already created
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in [UPDATING.md](https://github.com/apache/airflow/blob/main/UPDATING.md).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] rsg17 commented on pull request #21704: Add GCSToTrinoOperator
Posted by GitBox <gi...@apache.org>.
rsg17 commented on pull request #21704:
URL: https://github.com/apache/airflow/pull/21704#issuecomment-1047450308
Here is an updated version without the target_fields in the first row of the csv. I think all tests would pass for this.
I am still not sure why `fields = tuple(next(data))` failed with `TypeError: 'tuple' object is not an iterator`. I finally ended excluding it from the code.
cc - @eladkal, @uranusjr
--
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] rsg17 removed a comment on pull request #21704: Add GCSToTrinoOperator
Posted by GitBox <gi...@apache.org>.
rsg17 removed a comment on pull request #21704:
URL: https://github.com/apache/airflow/pull/21704#issuecomment-1048498886
Added a way to provide schema / fields.
--
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] rsg17 edited a comment on pull request #21704: Add GCSToTrinoOperator
Posted by GitBox <gi...@apache.org>.
rsg17 edited a comment on pull request #21704:
URL: https://github.com/apache/airflow/pull/21704#issuecomment-1050541006
@eladkal , @uranusjr : I have added alternate ways to provide `fields`. User can either provide them as a list or point to a json file on GCS that has the fields. I referred to how this was done for [gcs_to_bigquery](https://github.com/apache/airflow/blob/602abe8394fafe7de54df7e73af56de848cdf617/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py)
Do you think these will be enough?
--
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 merged pull request #21704: Add GCSToTrinoOperator
Posted by GitBox <gi...@apache.org>.
eladkal merged pull request #21704:
URL: https://github.com/apache/airflow/pull/21704
--
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] rsg17 commented on pull request #21704: Add GCSToTrinoOperator
Posted by GitBox <gi...@apache.org>.
rsg17 commented on pull request #21704:
URL: https://github.com/apache/airflow/pull/21704#issuecomment-1047944173
> But `fields` is gone entirely in your new commit. Is it intended?
Yes. It is intended because it does not work with the logic for `fields`.
I can add a different way to specify schema (maybe a separate json file); but not `fields`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] josh-fell commented on a change in pull request #21704: Add GCSToTrinoOperator
Posted by GitBox <gi...@apache.org>.
josh-fell commented on a change in pull request #21704:
URL: https://github.com/apache/airflow/pull/21704#discussion_r813975757
##########
File path: airflow/providers/trino/transfers/gcs_to_trino.py
##########
@@ -0,0 +1,118 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""This module contains Google Cloud Storage to Trino operator."""
+
+import csv
+import json
+from tempfile import NamedTemporaryFile
+from typing import TYPE_CHECKING, Iterable, Optional, Sequence, Union
+
+from airflow.models import BaseOperator
+from airflow.providers.google.cloud.hooks.gcs import GCSHook
+from airflow.providers.trino.hooks.trino import TrinoHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class GCSToTrinoOperator(BaseOperator):
Review comment:
Within operators and sensors you can specify `template_fields` which allow the values for those args to be Jinja templated and unlock some functionality with TaskFlow API. The Concepts docs have some [information on Jinja templating in operators](https://airflow.apache.org/docs/apache-airflow/stable/concepts/operators.html#jinja-templating).
So you'd want to think about how users might interact with this operator. Could they potentially use any of the [built-in Jinja templates](https://airflow.apache.org/docs/apache-airflow/stable/templates-ref.html) as part of the arg for certain parameters or even want to use an output from an upstream task as the value?
Storage buckets/containers and object names are pretty classic examples where Jinja templating is used frequently. Especially if you think about date-partitioned paths. This is why I was suggesting `source_bucket` and `source_object` as `template_fields`. But adding `trino_table` as an arg that can be templated can't hurt anything though.
--
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] uranusjr commented on a change in pull request #21704: Add GCSToTrinoOperator
Posted by GitBox <gi...@apache.org>.
uranusjr commented on a change in pull request #21704:
URL: https://github.com/apache/airflow/pull/21704#discussion_r810854496
##########
File path: airflow/providers/trino/transfers/gcs_to_trino.py
##########
@@ -0,0 +1,103 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""This module contains Google Cloud Storage to Trino operator."""
+
+import csv
+from tempfile import NamedTemporaryFile
+from typing import TYPE_CHECKING, Optional, Sequence, Union
+
+from airflow.models import BaseOperator
+from airflow.providers.google.cloud.hooks.gcs import GCSHook
+from airflow.providers.trino.hooks.trino import TrinoHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class GCSToTrinoOperator(BaseOperator):
+ """
+ Loads a csv file from Google Cloud Storage into a Trino table.
+ Assumptions:
+ 1. First row of the csv contains headers
+ 2. Trino table with requisite columns is already created
+
+ :param source_bucket: Source GCS bucket that contains the csv
+ :param source_object: csv file including the path
+ :param trino_table: trino table to upload the data
+ :param trino_conn_id: destination trino connection
+ :param gcp_conn_id: (Optional) The connection ID used to connect to Google Cloud and
+ interact with the Google Cloud Storage service.
+ :param delegate_to: The account to impersonate using domain-wide delegation of authority,
+ if any. For this to work, the service account making the request must have
+ domain-wide delegation enabled.
+ :param impersonation_chain: Optional service account to impersonate using short-term
+ credentials, or chained list of accounts required to get the access_token
+ of the last account in the list, which will be impersonated in the request.
+ If set as a string, the account must grant the originating account
+ the Service Account Token Creator IAM role.
+ If set as a sequence, the identities from the list must grant
+ Service Account Token Creator IAM role to the directly preceding identity, with first
+ account from the list granting this role to the originating account.
+ """
+
+ def __init__(
+ self,
+ *,
+ source_bucket: str,
+ source_object: str,
+ trino_table: str,
+ trino_conn_id: str = "trino_default",
+ gcp_conn_id: str = "google_cloud_default",
+ delegate_to: Optional[str] = None,
+ impersonation_chain: Optional[Union[str, Sequence[str]]] = None,
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.source_bucket = source_bucket
+ self.source_object = source_object
+ self.trino_table = trino_table
+ self.trino_conn_id = trino_conn_id
+ self.gcp_conn_id = gcp_conn_id
+ self.delegate_to = delegate_to
+ self.impersonation_chain = impersonation_chain
+
+ def execute(self, context: 'Context') -> None:
+ gcs_hook = GCSHook(
+ gcp_conn_id=self.gcp_conn_id,
+ delegate_to=self.delegate_to,
+ impersonation_chain=self.impersonation_chain,
+ )
+
+ trino_hook = TrinoHook(trino_conn_id=self.trino_conn_id)
+
+ with NamedTemporaryFile("w+") as temp_file:
+ self.log.info("Downloading data from %s", self.source_object)
+ gcs_hook.download(
+ bucket_name=self.source_bucket,
+ object_name=self.source_object,
+ filename=temp_file.name,
+ )
+
+ data = list(csv.reader(temp_file))
+ fields = tuple(data[0])
+ rows = []
+ for row in data[1:]:
+ rows.append(tuple(row))
Review comment:
> `TypeError: 'tuple' object is not an iterator`
This doesn’t sound right. The `next` call is on a `csv.reader()`, which is not a tuple. Did you mistype this as `next(tuple(data))`?
--
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] uranusjr commented on pull request #21704: Add GCSToTrinoOperator
Posted by GitBox <gi...@apache.org>.
uranusjr commented on pull request #21704:
URL: https://github.com/apache/airflow/pull/21704#issuecomment-1047466290
But `fields` is gone entirely in your new commit. Is it intended?
--
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] rsg17 commented on a change in pull request #21704: Add GCSToTrinoOperator
Posted by GitBox <gi...@apache.org>.
rsg17 commented on a change in pull request #21704:
URL: https://github.com/apache/airflow/pull/21704#discussion_r814487189
##########
File path: airflow/providers/trino/transfers/gcs_to_trino.py
##########
@@ -0,0 +1,118 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""This module contains Google Cloud Storage to Trino operator."""
+
+import csv
+import json
+from tempfile import NamedTemporaryFile
+from typing import TYPE_CHECKING, Iterable, Optional, Sequence, Union
+
+from airflow.models import BaseOperator
+from airflow.providers.google.cloud.hooks.gcs import GCSHook
+from airflow.providers.trino.hooks.trino import TrinoHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class GCSToTrinoOperator(BaseOperator):
Review comment:
Added `template_fields`
##########
File path: airflow/providers/trino/transfers/gcs_to_trino.py
##########
@@ -0,0 +1,118 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""This module contains Google Cloud Storage to Trino operator."""
+
+import csv
+import json
+from tempfile import NamedTemporaryFile
+from typing import TYPE_CHECKING, Iterable, Optional, Sequence, Union
+
+from airflow.models import BaseOperator
+from airflow.providers.google.cloud.hooks.gcs import GCSHook
+from airflow.providers.trino.hooks.trino import TrinoHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class GCSToTrinoOperator(BaseOperator):
Review comment:
Added `template_fields`. Thank you!!
--
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] rsg17 commented on a change in pull request #21704: Add GCSToTrinoOperator
Posted by GitBox <gi...@apache.org>.
rsg17 commented on a change in pull request #21704:
URL: https://github.com/apache/airflow/pull/21704#discussion_r810858212
##########
File path: airflow/providers/trino/transfers/gcs_to_trino.py
##########
@@ -0,0 +1,103 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""This module contains Google Cloud Storage to Trino operator."""
+
+import csv
+from tempfile import NamedTemporaryFile
+from typing import TYPE_CHECKING, Optional, Sequence, Union
+
+from airflow.models import BaseOperator
+from airflow.providers.google.cloud.hooks.gcs import GCSHook
+from airflow.providers.trino.hooks.trino import TrinoHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class GCSToTrinoOperator(BaseOperator):
+ """
+ Loads a csv file from Google Cloud Storage into a Trino table.
+ Assumptions:
+ 1. First row of the csv contains headers
+ 2. Trino table with requisite columns is already created
+
+ :param source_bucket: Source GCS bucket that contains the csv
+ :param source_object: csv file including the path
+ :param trino_table: trino table to upload the data
+ :param trino_conn_id: destination trino connection
+ :param gcp_conn_id: (Optional) The connection ID used to connect to Google Cloud and
+ interact with the Google Cloud Storage service.
+ :param delegate_to: The account to impersonate using domain-wide delegation of authority,
+ if any. For this to work, the service account making the request must have
+ domain-wide delegation enabled.
+ :param impersonation_chain: Optional service account to impersonate using short-term
+ credentials, or chained list of accounts required to get the access_token
+ of the last account in the list, which will be impersonated in the request.
+ If set as a string, the account must grant the originating account
+ the Service Account Token Creator IAM role.
+ If set as a sequence, the identities from the list must grant
+ Service Account Token Creator IAM role to the directly preceding identity, with first
+ account from the list granting this role to the originating account.
+ """
+
+ def __init__(
+ self,
+ *,
+ source_bucket: str,
+ source_object: str,
+ trino_table: str,
+ trino_conn_id: str = "trino_default",
+ gcp_conn_id: str = "google_cloud_default",
+ delegate_to: Optional[str] = None,
+ impersonation_chain: Optional[Union[str, Sequence[str]]] = None,
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.source_bucket = source_bucket
+ self.source_object = source_object
+ self.trino_table = trino_table
+ self.trino_conn_id = trino_conn_id
+ self.gcp_conn_id = gcp_conn_id
+ self.delegate_to = delegate_to
+ self.impersonation_chain = impersonation_chain
+
+ def execute(self, context: 'Context') -> None:
+ gcs_hook = GCSHook(
+ gcp_conn_id=self.gcp_conn_id,
+ delegate_to=self.delegate_to,
+ impersonation_chain=self.impersonation_chain,
+ )
+
+ trino_hook = TrinoHook(trino_conn_id=self.trino_conn_id)
+
+ with NamedTemporaryFile("w+") as temp_file:
+ self.log.info("Downloading data from %s", self.source_object)
+ gcs_hook.download(
+ bucket_name=self.source_bucket,
+ object_name=self.source_object,
+ filename=temp_file.name,
+ )
+
+ data = list(csv.reader(temp_file))
+ fields = tuple(data[0])
+ rows = []
+ for row in data[1:]:
+ rows.append(tuple(row))
Review comment:
No - I did not mistype that. I was surprised it failed too!
I have pushed the change (I expect CI would fail).
--
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] rsg17 commented on pull request #21704: Add GCSToTrinoOperator
Posted by GitBox <gi...@apache.org>.
rsg17 commented on pull request #21704:
URL: https://github.com/apache/airflow/pull/21704#issuecomment-1046507285
r? @eladkal
Here is the PR for GCSToTrinoOperator that we discussed on the GCSToPrestoOperator PR.
--
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 #21704: Add GCSToTrinoOperator
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #21704:
URL: https://github.com/apache/airflow/pull/21704#issuecomment-1050551954
The PR is likely OK to be merged with just subset of tests for default Python and Database versions without running the full matrix of tests, because it does not modify the core of Airflow. If the committers decide that the full tests matrix is needed, they will add the label 'full tests needed'. Then you should rebase to the latest main or amend the last commit of the PR, and push it with --force-with-lease.
--
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] rsg17 commented on a change in pull request #21704: Add GCSToTrinoOperator
Posted by GitBox <gi...@apache.org>.
rsg17 commented on a change in pull request #21704:
URL: https://github.com/apache/airflow/pull/21704#discussion_r810852970
##########
File path: airflow/providers/trino/transfers/gcs_to_trino.py
##########
@@ -0,0 +1,103 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""This module contains Google Cloud Storage to Trino operator."""
+
+import csv
+from tempfile import NamedTemporaryFile
+from typing import TYPE_CHECKING, Optional, Sequence, Union
+
+from airflow.models import BaseOperator
+from airflow.providers.google.cloud.hooks.gcs import GCSHook
+from airflow.providers.trino.hooks.trino import TrinoHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class GCSToTrinoOperator(BaseOperator):
+ """
+ Loads a csv file from Google Cloud Storage into a Trino table.
+ Assumptions:
+ 1. First row of the csv contains headers
+ 2. Trino table with requisite columns is already created
+
+ :param source_bucket: Source GCS bucket that contains the csv
+ :param source_object: csv file including the path
+ :param trino_table: trino table to upload the data
+ :param trino_conn_id: destination trino connection
+ :param gcp_conn_id: (Optional) The connection ID used to connect to Google Cloud and
+ interact with the Google Cloud Storage service.
+ :param delegate_to: The account to impersonate using domain-wide delegation of authority,
+ if any. For this to work, the service account making the request must have
+ domain-wide delegation enabled.
+ :param impersonation_chain: Optional service account to impersonate using short-term
+ credentials, or chained list of accounts required to get the access_token
+ of the last account in the list, which will be impersonated in the request.
+ If set as a string, the account must grant the originating account
+ the Service Account Token Creator IAM role.
+ If set as a sequence, the identities from the list must grant
+ Service Account Token Creator IAM role to the directly preceding identity, with first
+ account from the list granting this role to the originating account.
+ """
+
+ def __init__(
+ self,
+ *,
+ source_bucket: str,
+ source_object: str,
+ trino_table: str,
+ trino_conn_id: str = "trino_default",
+ gcp_conn_id: str = "google_cloud_default",
+ delegate_to: Optional[str] = None,
+ impersonation_chain: Optional[Union[str, Sequence[str]]] = None,
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.source_bucket = source_bucket
+ self.source_object = source_object
+ self.trino_table = trino_table
+ self.trino_conn_id = trino_conn_id
+ self.gcp_conn_id = gcp_conn_id
+ self.delegate_to = delegate_to
+ self.impersonation_chain = impersonation_chain
+
+ def execute(self, context: 'Context') -> None:
+ gcs_hook = GCSHook(
+ gcp_conn_id=self.gcp_conn_id,
+ delegate_to=self.delegate_to,
+ impersonation_chain=self.impersonation_chain,
+ )
+
+ trino_hook = TrinoHook(trino_conn_id=self.trino_conn_id)
+
+ with NamedTemporaryFile("w+") as temp_file:
+ self.log.info("Downloading data from %s", self.source_object)
+ gcs_hook.download(
+ bucket_name=self.source_bucket,
+ object_name=self.source_object,
+ filename=temp_file.name,
+ )
+
+ data = list(csv.reader(temp_file))
+ fields = tuple(data[0])
+ rows = []
+ for row in data[1:]:
+ rows.append(tuple(row))
Review comment:
@uranusjr
`fields = tuple(next(data))` fails with `TypeError: 'tuple' object is not an iterator` when I run the unit-test.
The line flagged from the test file is `op.execute(None)`
I am not really sure why this fails. The error indicates I am passing in a tuple in place of a csv-reader. But, given the line from the test file at which the failure occurs; I am not sure how I can pass in a csv-reader.
--
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 pull request #21704: Add GCSToTrinoOperator
Posted by GitBox <gi...@apache.org>.
eladkal commented on pull request #21704:
URL: https://github.com/apache/airflow/pull/21704#issuecomment-1046887543
Tests are failing:
```
self = <Task(GCSToTrinoOperator): test_gcs_to_trino>, context = None
def execute(self, context: 'Context') -> None:
gcs_hook = GCSHook(
gcp_conn_id=self.gcp_conn_id,
delegate_to=self.delegate_to,
impersonation_chain=self.impersonation_chain,
)
trino_hook = TrinoHook(trino_conn_id=self.trino_conn_id)
with NamedTemporaryFile("w+") as temp_file:
self.log.info("Downloading data from %s", self.source_object)
gcs_hook.download(
bucket_name=self.source_bucket,
object_name=self.source_object,
filename=temp_file.name,
)
data = csv.reader(temp_file)
> fields = tuple(next(data))
E TypeError: 'tuple' object is not an iterator
airflow/providers/trino/transfers/gcs_to_trino.py:97: TypeError
```
--
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] rsg17 commented on pull request #21704: Add GCSToTrinoOperator
Posted by GitBox <gi...@apache.org>.
rsg17 commented on pull request #21704:
URL: https://github.com/apache/airflow/pull/21704#issuecomment-1047046996
> Tests are failing:
> ```
>
> self = <Task(GCSToTrinoOperator): test_gcs_to_trino>, context = None
>
> def execute(self, context: 'Context') -> None:
> gcs_hook = GCSHook(
> gcp_conn_id=self.gcp_conn_id,
> delegate_to=self.delegate_to,
> impersonation_chain=self.impersonation_chain,
> )
>
> trino_hook = TrinoHook(trino_conn_id=self.trino_conn_id)
>
> with NamedTemporaryFile("w+") as temp_file:
> self.log.info("Downloading data from %s", self.source_object)
> gcs_hook.download(
> bucket_name=self.source_bucket,
> object_name=self.source_object,
> filename=temp_file.name,
> )
>
> data = csv.reader(temp_file)
> > fields = tuple(next(data))
> E TypeError: 'tuple' object is not an iterator
>
> airflow/providers/trino/transfers/gcs_to_trino.py:97: TypeError
> ```
Yes. I made the change based on @uranusjr suggestion. This logic is more efficient than reading in the whole csv at once; but it causes the unit tests to fail.
I pushed the change upstream so that I can get suggestions on why this happens.
--
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] uranusjr commented on a change in pull request #21704: Add GCSToTrinoOperator
Posted by GitBox <gi...@apache.org>.
uranusjr commented on a change in pull request #21704:
URL: https://github.com/apache/airflow/pull/21704#discussion_r810806292
##########
File path: airflow/providers/trino/transfers/gcs_to_trino.py
##########
@@ -0,0 +1,103 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""This module contains Google Cloud Storage to Trino operator."""
+
+import csv
+from tempfile import NamedTemporaryFile
+from typing import TYPE_CHECKING, Optional, Sequence, Union
+
+from airflow.models import BaseOperator
+from airflow.providers.google.cloud.hooks.gcs import GCSHook
+from airflow.providers.trino.hooks.trino import TrinoHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class GCSToTrinoOperator(BaseOperator):
+ """
+ Loads a csv file from Google Cloud Storage into a Trino table.
+ Assumptions:
+ 1. First row of the csv contains headers
+ 2. Trino table with requisite columns is already created
+
+ :param source_bucket: Source GCS bucket that contains the csv
+ :param source_object: csv file including the path
+ :param trino_table: trino table to upload the data
+ :param trino_conn_id: destination trino connection
+ :param gcp_conn_id: (Optional) The connection ID used to connect to Google Cloud and
+ interact with the Google Cloud Storage service.
+ :param delegate_to: The account to impersonate using domain-wide delegation of authority,
+ if any. For this to work, the service account making the request must have
+ domain-wide delegation enabled.
+ :param impersonation_chain: Optional service account to impersonate using short-term
+ credentials, or chained list of accounts required to get the access_token
+ of the last account in the list, which will be impersonated in the request.
+ If set as a string, the account must grant the originating account
+ the Service Account Token Creator IAM role.
+ If set as a sequence, the identities from the list must grant
+ Service Account Token Creator IAM role to the directly preceding identity, with first
+ account from the list granting this role to the originating account.
+ """
+
+ def __init__(
+ self,
+ *,
+ source_bucket: str,
+ source_object: str,
+ trino_table: str,
+ trino_conn_id: str = "trino_default",
+ gcp_conn_id: str = "google_cloud_default",
+ delegate_to: Optional[str] = None,
+ impersonation_chain: Optional[Union[str, Sequence[str]]] = None,
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.source_bucket = source_bucket
+ self.source_object = source_object
+ self.trino_table = trino_table
+ self.trino_conn_id = trino_conn_id
+ self.gcp_conn_id = gcp_conn_id
+ self.delegate_to = delegate_to
+ self.impersonation_chain = impersonation_chain
+
+ def execute(self, context: 'Context') -> None:
+ gcs_hook = GCSHook(
+ gcp_conn_id=self.gcp_conn_id,
+ delegate_to=self.delegate_to,
+ impersonation_chain=self.impersonation_chain,
+ )
+
+ trino_hook = TrinoHook(trino_conn_id=self.trino_conn_id)
+
+ with NamedTemporaryFile("w+") as temp_file:
+ self.log.info("Downloading data from %s", self.source_object)
+ gcs_hook.download(
+ bucket_name=self.source_bucket,
+ object_name=self.source_object,
+ filename=temp_file.name,
+ )
+
+ data = list(csv.reader(temp_file))
+ fields = tuple(data[0])
+ rows = []
+ for row in data[1:]:
+ rows.append(tuple(row))
Review comment:
```suggestion
data = csv.reader(temp_file)
fields = tuple(next(data))
rows = (tuple(row) for row in data)
```
This saves significant memory usage when the downloaded file is large.
--
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] rsg17 commented on pull request #21704: Add GCSToTrinoOperator
Posted by GitBox <gi...@apache.org>.
rsg17 commented on pull request #21704:
URL: https://github.com/apache/airflow/pull/21704#issuecomment-1050541006
@eladkal , @uranusjr : I have added alternate ways to provide `fields`. User can either provide them as a list or point to a json file on GCS that has the fields.
--
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] rsg17 commented on pull request #21704: Add GCSToTrinoOperator
Posted by GitBox <gi...@apache.org>.
rsg17 commented on pull request #21704:
URL: https://github.com/apache/airflow/pull/21704#issuecomment-1048498886
Added a way to provide schema / fields.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] josh-fell commented on a change in pull request #21704: Add GCSToTrinoOperator
Posted by GitBox <gi...@apache.org>.
josh-fell commented on a change in pull request #21704:
URL: https://github.com/apache/airflow/pull/21704#discussion_r813056029
##########
File path: airflow/providers/trino/transfers/gcs_to_trino.py
##########
@@ -0,0 +1,118 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""This module contains Google Cloud Storage to Trino operator."""
+
+import csv
+import json
+from tempfile import NamedTemporaryFile
+from typing import TYPE_CHECKING, Iterable, Optional, Sequence, Union
+
+from airflow.models import BaseOperator
+from airflow.providers.google.cloud.hooks.gcs import GCSHook
+from airflow.providers.trino.hooks.trino import TrinoHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class GCSToTrinoOperator(BaseOperator):
Review comment:
Are there are args for this operator that could dynamically generated and should be `template_fields`? I'm thinking `source_bucket`, `source_object`, and `trino_table`(?) might be good candidates here.
And, on a related note, probably worth also adding the same `template_fields` to `GCSToPrestoOperator` in a separate PR too.
--
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] rsg17 commented on a change in pull request #21704: Add GCSToTrinoOperator
Posted by GitBox <gi...@apache.org>.
rsg17 commented on a change in pull request #21704:
URL: https://github.com/apache/airflow/pull/21704#discussion_r813514536
##########
File path: airflow/providers/trino/transfers/gcs_to_trino.py
##########
@@ -0,0 +1,118 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""This module contains Google Cloud Storage to Trino operator."""
+
+import csv
+import json
+from tempfile import NamedTemporaryFile
+from typing import TYPE_CHECKING, Iterable, Optional, Sequence, Union
+
+from airflow.models import BaseOperator
+from airflow.providers.google.cloud.hooks.gcs import GCSHook
+from airflow.providers.trino.hooks.trino import TrinoHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class GCSToTrinoOperator(BaseOperator):
Review comment:
Can you give an example of template fields / dynamically generated fields?
I can make the change here and for the presto operator after that
--
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