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 2020/05/15 16:01:20 UTC
[GitHub] [airflow] feluelle opened a new pull request #8877: [WIP] Add S3ToRedshift example dag and system test
feluelle opened a new pull request #8877:
URL: https://github.com/apache/airflow/pull/8877
This PR / change adds an example dag and tests it (=system test).
NOTE: Everything except the Redshift Cluster will be _provided_. So the Redshift Cluster needs to exist and so must the connection to it. I set env variable `AIRFLOW_CONN_REDSHIFT_DEFAULT` to do that.
---
Make sure to mark the boxes below before creating PR: [x]
- [x] Description above provides context of the change
- [x] Unit tests coverage for changes (not needed for documentation changes)
- [x] Target Github ISSUE in description if exists
- [x] Commits follow "[How to write a good git commit message](http://chris.beams.io/posts/git-commit/)"
- [ ] Relevant documentation is updated including usage instructions.
- [x] I will engage committers as explained in [Contribution Workflow Example](https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#contribution-workflow-example).
---
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/master/UPDATING.md).
Read the [Pull Request Guidelines](https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#pull-request-guidelines) for more information.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #8877: Add S3ToRedshift example dag and system test
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #8877:
URL: https://github.com/apache/airflow/pull/8877#discussion_r437309397
##########
File path: tests/providers/amazon/aws/operators/test_s3_to_redshift_system.py
##########
@@ -0,0 +1,44 @@
+#
+# 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.
+import os
+
+import pytest
+
+from airflow.models import Connection
+from airflow.utils import db
+from tests.test_utils import AIRFLOW_MAIN_FOLDER
+from tests.test_utils.amazon_system_helpers import AWS_DAG_FOLDER, AmazonSystemTest
+from tests.test_utils.terraform import Terraform
+
+
+@pytest.mark.backend("mysql", "postgres")
+class TestS3ToRedshiftExampleDags(AmazonSystemTest, Terraform):
+ TERRAFORM_DIR = os.path.join(
+ AIRFLOW_MAIN_FOLDER, "tests", "providers", "amazon", "aws", "infrastructure", "example_s3_to_redshift"
+ )
+
+ def setUp(self) -> None:
+ super().setUp()
+ host, port = self.get_tf_output("redshift_endpoint").split(':')
+ schema = self.get_tf_output("redshift_database_name"),
+ login = self.get_tf_output("redshift_master_username"),
+ password = self.get_tf_output("redshift_master_password")
+ db.merge_conn(Connection("redshift_default", "postgres", host, login, password, schema, port))
Review comment:
Nope. That's a TODO. Thanks 👍
##########
File path: airflow/providers/amazon/aws/example_dags/example_s3_to_redshift.py
##########
@@ -41,13 +42,26 @@
schedule_interval=None,
tags=['example']
) as dag:
+ preparation__task_load_sample_data_to_s3 = PythonOperator(
+ python_callable=lambda: S3Hook().load_string("0,Airflow", f'{S3_KEY}/{REDSHIFT_TABLE}', S3_BUCKET),
+ task_id='preparation__load_sample_data_to_s3'
+ )
+ preparation__task_create_table = PostgresOperator(
+ sql=f'CREATE TABLE {REDSHIFT_TABLE}(Id int, Name varchar)',
+ postgres_conn_id='redshift_default',
+ task_id='preparation__create_table'
+ )
Review comment:
You are correct. But since I am `terraform destroy`ing the whole infra being used in `tearDown`. I don't need to remove the s3 object and the redshift table, do I?
##########
File path: airflow/providers/amazon/aws/example_dags/example_s3_to_redshift.py
##########
@@ -41,13 +42,26 @@
schedule_interval=None,
tags=['example']
) as dag:
+ preparation__task_load_sample_data_to_s3 = PythonOperator(
+ python_callable=lambda: S3Hook().load_string("0,Airflow", f'{S3_KEY}/{REDSHIFT_TABLE}', S3_BUCKET),
+ task_id='preparation__load_sample_data_to_s3'
+ )
+ preparation__task_create_table = PostgresOperator(
+ sql=f'CREATE TABLE {REDSHIFT_TABLE}(Id int, Name varchar)',
+ postgres_conn_id='redshift_default',
+ task_id='preparation__create_table'
+ )
Review comment:
You are correct. But since I am `terraform destroy`ing the whole infra being used in `tearDown` I don't need to remove the s3 object and the redshift table, do I?
##########
File path: tests/test_utils/terraform.py
##########
@@ -0,0 +1,36 @@
+# 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 tests.test_utils.system_tests_class import SystemTest
+
+
+class Terraform(SystemTest):
+ TERRAFORM_DIR: str
+
+ def setUp(self) -> None:
+ self.execute_cmd(["terraform", "init", "-input=false", self.TERRAFORM_DIR])
+ self.execute_cmd(["terraform", "plan", "-input=false", self.TERRAFORM_DIR])
+ self.execute_cmd(["terraform", "apply", "-input=false", "-auto-approve", self.TERRAFORM_DIR])
+
+ def get_tf_output(self, name):
+ output = self.check_output(["terraform", "output", name]).decode('utf-8').replace("\r\n", "")
+ self.log.info(output)
+ return output
+
+ def tearDown(self) -> None:
+ self.execute_cmd(["terraform", "plan", "-destroy", "-input=false", self.TERRAFORM_DIR])
+ self.execute_cmd(["terraform", "destroy", "-input=false", "-auto-approve", self.TERRAFORM_DIR])
Review comment:
I checked this out, but didn't find that really useful to use an **extra package** only to wrap bash inside python. And it also [seems](https://github.com/beelit94/python-terraform/issues/85) to not be compatible to the latest terraform 0.12.x releases.
##########
File path: airflow/providers/amazon/aws/example_dags/example_s3_to_redshift.py
##########
@@ -41,13 +42,26 @@
schedule_interval=None,
tags=['example']
) as dag:
+ preparation__task_load_sample_data_to_s3 = PythonOperator(
+ python_callable=lambda: S3Hook().load_string("0,Airflow", f'{S3_KEY}/{REDSHIFT_TABLE}', S3_BUCKET),
+ task_id='preparation__load_sample_data_to_s3'
+ )
+ preparation__task_create_table = PostgresOperator(
+ sql=f'CREATE TABLE {REDSHIFT_TABLE}(Id int, Name varchar)',
+ postgres_conn_id='redshift_default',
+ task_id='preparation__create_table'
+ )
Review comment:
I do need to remove it, you are right. 👍 It should not depend on the `tearDown`.
And I will also make the setup steps idempotent. 👍
##########
File path: airflow/providers/amazon/aws/example_dags/example_s3_to_redshift.py
##########
@@ -0,0 +1,53 @@
+# 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 is an example dag for using `S3ToRedshiftTransfer` to copy a S3 key into a Redshift table.
+"""
+
+from os import getenv
+
+from airflow import DAG
+from airflow.providers.amazon.aws.operators.s3_to_redshift import S3ToRedshiftTransfer
+from airflow.utils.dates import days_ago
+
+# [START howto_operator_s3_to_redshift_env_variables]
+S3_BUCKET = getenv("S3_BUCKET", "bucket")
+S3_KEY = getenv("S3_KEY", "key")
Review comment:
But how do we then define defaults in case these `var.`s are not set? Or should we make all vars required then?
##########
File path: tests/test_utils/terraform.py
##########
@@ -0,0 +1,36 @@
+# 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 tests.test_utils.system_tests_class import SystemTest
+
+
+class Terraform(SystemTest):
+ TERRAFORM_DIR: str
+
+ def setUp(self) -> None:
+ self.execute_cmd(["terraform", "init", "-input=false", self.TERRAFORM_DIR])
+ self.execute_cmd(["terraform", "plan", "-input=false", self.TERRAFORM_DIR])
+ self.execute_cmd(["terraform", "apply", "-input=false", "-auto-approve", self.TERRAFORM_DIR])
+
+ def get_tf_output(self, name):
+ output = self.check_output(["terraform", "output", name]).decode('utf-8').replace("\r\n", "")
+ self.log.info(output)
+ return output
+
+ def tearDown(self) -> None:
+ self.execute_cmd(["terraform", "plan", "-destroy", "-input=false", self.TERRAFORM_DIR])
+ self.execute_cmd(["terraform", "destroy", "-input=false", "-auto-approve", self.TERRAFORM_DIR])
Review comment:
I could do that.
The real issue I am having with this, is also that `super().setUp()` and `super().tearDown()` are not explicit enough. Terraform is very much hidden behind them. You can easily forget to call those when you overwrite those which for `super().tearDown()` could be dramatically. But maybe we could add a static check that checks 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] xinbinhuang commented on a change in pull request #8877: Add S3ToRedshift example dag and system test
Posted by GitBox <gi...@apache.org>.
xinbinhuang commented on a change in pull request #8877:
URL: https://github.com/apache/airflow/pull/8877#discussion_r426189138
##########
File path: tests/providers/amazon/aws/operators/test_s3_to_redshift_system.py
##########
@@ -0,0 +1,62 @@
+#
+# 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 tempfile import NamedTemporaryFile
+
+import pytest
+
+from airflow.providers.amazon.aws.example_dags.example_s3_to_redshift import (
+ REDSHIFT_SCHEMA, REDSHIFT_TABLE, S3_BUCKET, S3_KEY,
+)
+from tests.test_utils.amazon_system_helpers import (
+ AWS_DAG_FOLDER, AmazonSystemTest, provide_aws_context, provide_aws_s3_bucket, provide_aws_s3_key,
+)
+from tests.test_utils.postgres_system_helpers import provide_postgres_schema, provide_postgres_table
+
+
+@pytest.fixture
+def provide_s3_bucket_with_key():
+ with provide_aws_s3_bucket(name=S3_BUCKET) as s3_bucket:
+ with NamedTemporaryFile() as file:
Review comment:
```suggestion
with provide_aws_s3_bucket(name=S3_BUCKET) as s3_bucket, NamedTemporaryFile() as file:
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #8877: [WIP] Add S3ToRedshift example dag and system test
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #8877:
URL: https://github.com/apache/airflow/pull/8877#discussion_r425901097
##########
File path: tests/test_utils/postgres_system_helpers.py
##########
@@ -0,0 +1,97 @@
+# 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 contextlib import contextmanager
+from typing import List
+
+import pytest
+
+from tests.test_utils.system_tests_class import SystemTest
+from tests.utils.logging_command_executor import get_executor
+
+
+@contextmanager
+def provide_postgres_schema(name):
+ PostgresSystemTest.create_schema(name)
+ yield name
+ PostgresSystemTest.delete_schema(name)
+
+
+@contextmanager
+def provide_postgres_table(schema_name, table_name, columns=None):
+ PostgresSystemTest.create_table(schema_name, table_name, columns)
+ yield
+ PostgresSystemTest.delete_table(schema_name, table_name)
+
+
+@pytest.mark.system("postgres")
+class PostgresSystemTest(SystemTest):
+
+ @classmethod
+ def execute_with_ctx(cls, cmd: str):
+ """
+ Executes command with postgres (psql cli) context.
+ """
+ executor = get_executor()
+ from os import getenv
+ connection_string = getenv("AIRFLOW_CONN_POSTGRES_DEFAULT")
+ auth = ["psql", connection_string]
+ executor.execute_cmd(cmd=auth + ["-c", cmd])
Review comment:
> from os import getenv
> connection_string = getenv("AIRFLOW_CONN_POSTGRES_DEFAULT")
Getting the connection through a env variable that needs to be defined :/ Is there a better way to do that?
> auth = ["psql", connection_string]
> executor.execute_cmd(cmd=auth + ["-c", cmd])
I don't like that part. I want to authenticate the context / the `psql` commands - that was only thing that worked.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mustafagok commented on a change in pull request #8877: Add S3ToRedshift example dag and system test
Posted by GitBox <gi...@apache.org>.
mustafagok commented on a change in pull request #8877:
URL: https://github.com/apache/airflow/pull/8877#discussion_r426184251
##########
File path: tests/test_utils/postgres_system_helpers.py
##########
@@ -0,0 +1,97 @@
+# 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 contextlib import contextmanager
+from typing import List
+
+import pytest
+
+from tests.test_utils.system_tests_class import SystemTest
+from tests.utils.logging_command_executor import get_executor
+
+
+@contextmanager
+def provide_postgres_schema(name):
+ PostgresSystemTest.create_schema(name)
+ yield name
+ PostgresSystemTest.delete_schema(name)
+
+
+@contextmanager
+def provide_postgres_table(schema_name, table_name, columns=None):
+ PostgresSystemTest.create_table(schema_name, table_name, columns)
+ yield
+ PostgresSystemTest.delete_table(schema_name, table_name)
+
+
+@pytest.mark.system("postgres")
+class PostgresSystemTest(SystemTest):
+
+ @classmethod
+ def execute_with_ctx(cls, cmd: str):
+ """
+ Executes command with postgres (psql cli) context.
+ """
+ executor = get_executor()
+ from os import getenv
+ connection_string = getenv("AIRFLOW_CONN_POSTGRES_DEFAULT")
+ auth = ["psql", connection_string]
+ executor.execute_cmd(cmd=auth + ["-c", cmd])
Review comment:
I have created connection in setUp, see: #8888
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #8877: Add S3ToRedshift example dag and system test
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #8877:
URL: https://github.com/apache/airflow/pull/8877#discussion_r426180536
##########
File path: airflow/providers/amazon/aws/example_dags/example_s3_to_redshift.py
##########
@@ -0,0 +1,53 @@
+# 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 is an example dag for using `S3ToRedshiftTransfer` to copy a S3 key into a Redshift table.
+"""
+
+from os import getenv
+
+from airflow import DAG
+from airflow.providers.amazon.aws.operators.s3_to_redshift import S3ToRedshiftTransfer
+from airflow.utils.dates import days_ago
+
+# [START howto_operator_s3_to_redshift_env_variables]
+S3_BUCKET = getenv("S3_BUCKET", "bucket")
+S3_KEY = getenv("S3_KEY", "key")
Review comment:
We have this in all of our system tests. So that we can easily pass them to the example dags. See https://github.com/apache/airflow/blob/master/TESTING.rst#environment-for-system-tests
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #8877: Add S3ToRedshift example dag and system test
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #8877:
URL: https://github.com/apache/airflow/pull/8877#discussion_r426180306
##########
File path: tests/test_utils/amazon_system_helpers.py
##########
@@ -82,6 +89,31 @@ def delete_aws_s3_bucket(cls, name: str) -> None:
cmd = ["aws", "s3api", "delete-bucket", "--bucket", name]
cls.execute_with_ctx(cmd)
+ @classmethod
+ def create_aws_s3_key(cls, bucket_name: str, key_name: str, body: Optional[str] = None) -> None:
+ """
+ Puts an object into the aws bucket.
+
+ :param bucket_name: name of the bucket
+ :param key_name: name of the key to add
+ :param body: optional content of the key to add
+ """
+ cmd = ["aws", "s3api", "put-object", "--bucket", bucket_name, "--key", key_name]
Review comment:
Also some function we don't have in our hooks or operators. We then would need to mix hook, operator and boto3 calls. I think it would get confusing / different abstraction layers.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #8877: Add S3ToRedshift example dag and system test
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #8877:
URL: https://github.com/apache/airflow/pull/8877#discussion_r426175319
##########
File path: tests/test_utils/amazon_system_helpers.py
##########
@@ -82,6 +89,31 @@ def delete_aws_s3_bucket(cls, name: str) -> None:
cmd = ["aws", "s3api", "delete-bucket", "--bucket", name]
cls.execute_with_ctx(cmd)
+ @classmethod
+ def create_aws_s3_key(cls, bucket_name: str, key_name: str, body: Optional[str] = None) -> None:
+ """
+ Puts an object into the aws bucket.
+
+ :param bucket_name: name of the bucket
+ :param key_name: name of the key to add
+ :param body: optional content of the key to add
+ """
+ cmd = ["aws", "s3api", "put-object", "--bucket", bucket_name, "--key", key_name]
Review comment:
Ummm. We have boto installed, let's use that instead of shelling out
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on pull request #8877: Add S3ToRedshift example dag and system test
Posted by GitBox <gi...@apache.org>.
feluelle commented on pull request #8877:
URL: https://github.com/apache/airflow/pull/8877#issuecomment-629680173
Ready for review. PTAL @ashb @potiuk
Maybe @xinbinhuang @mustafagok @baolsen if you have time :)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on pull request #8877: Add S3ToRedshift example dag and system test
Posted by GitBox <gi...@apache.org>.
feluelle commented on pull request #8877:
URL: https://github.com/apache/airflow/pull/8877#issuecomment-639691781
My variables which are not committed and I used for testing are:
_variables.env_
```
S3_BUCKET=test-airflow-bucket-${AIRFLOW_BREEZE_UNIQUE_SUFFIX}
TF_VAR_s3_bucket=${S3_BUCKET}
TF_VAR_redshift_cluster_identifier=test-airflow-cluster-${AIRFLOW_BREEZE_UNIQUE_SUFFIX}
HOST_AIRFLOW_SOURCES=/Users/feluelle/PycharmProjects/incubator-airflow
```
and _terraform.tfvars_
```
aws_region = "eu-central-1"
redshift_database_name = "dev"
redshift_master_username = "master"
redshift_master_password = "1Two3Four"
```
I added _terraform.tfvars_ to .gitignore, because I was not sure if we should commit it since it contains "sensitive information" eventhough we only use it for testing.
The full example dag takes about 7 mins to run.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk merged pull request #8877: Add S3ToRedshift example dag and system test
Posted by GitBox <gi...@apache.org>.
potiuk merged pull request #8877:
URL: https://github.com/apache/airflow/pull/8877
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #8877: [WIP] Add S3ToRedshift example dag and system test
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #8877:
URL: https://github.com/apache/airflow/pull/8877#discussion_r425906307
##########
File path: tests/providers/amazon/aws/operators/test_s3_to_redshift_system.py
##########
@@ -0,0 +1,62 @@
+#
+# 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 tempfile import NamedTemporaryFile
+
+import pytest
+
+from airflow.providers.amazon.aws.example_dags.example_s3_to_redshift import (
+ REDSHIFT_SCHEMA, REDSHIFT_TABLE, S3_BUCKET, S3_KEY,
+)
+from tests.test_utils.amazon_system_helpers import (
+ AWS_DAG_FOLDER, AmazonSystemTest, provide_aws_context, provide_aws_s3_bucket, provide_aws_s3_key,
+)
+from tests.test_utils.postgres_system_helpers import provide_postgres_schema, provide_postgres_table
+
+
+@pytest.fixture
+def provide_s3_bucket_with_key():
+ with provide_aws_s3_bucket(name=S3_BUCKET, region='eu-central-1') as s3_bucket:
Review comment:
I needed to add the `region` parameter so the s3 bucket will be created in that region. Because otherwise it would be in `us` and my existing cluster is in `eu`. The `COPY` command does not work across regions. :/
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #8877: Add S3ToRedshift example dag and system test
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #8877:
URL: https://github.com/apache/airflow/pull/8877#discussion_r426519274
##########
File path: tests/test_utils/postgres_system_helpers.py
##########
@@ -0,0 +1,97 @@
+# 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 contextlib import contextmanager
+from typing import List
+
+import pytest
+
+from tests.test_utils.system_tests_class import SystemTest
+from tests.utils.logging_command_executor import get_executor
+
+
+@contextmanager
+def provide_postgres_schema(name):
+ PostgresSystemTest.create_schema(name)
+ yield name
+ PostgresSystemTest.delete_schema(name)
+
+
+@contextmanager
+def provide_postgres_table(schema_name, table_name, columns=None):
+ PostgresSystemTest.create_table(schema_name, table_name, columns)
+ yield
+ PostgresSystemTest.delete_table(schema_name, table_name)
+
+
+@pytest.mark.system("postgres")
+class PostgresSystemTest(SystemTest):
+
+ @classmethod
+ def execute_with_ctx(cls, cmd: str):
+ """
+ Executes command with postgres (psql cli) context.
+ """
+ executor = get_executor()
+ from os import getenv
+ connection_string = getenv("AIRFLOW_CONN_POSTGRES_DEFAULT")
+ auth = ["psql", connection_string]
+ executor.execute_cmd(cmd=auth + ["-c", cmd])
Review comment:
Same as above.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #8877: Add S3ToRedshift example dag and system test
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #8877:
URL: https://github.com/apache/airflow/pull/8877#discussion_r426180163
##########
File path: tests/test_utils/amazon_system_helpers.py
##########
@@ -82,6 +89,31 @@ def delete_aws_s3_bucket(cls, name: str) -> None:
cmd = ["aws", "s3api", "delete-bucket", "--bucket", name]
cls.execute_with_ctx(cmd)
+ @classmethod
+ def create_aws_s3_key(cls, bucket_name: str, key_name: str, body: Optional[str] = None) -> None:
+ """
+ Puts an object into the aws bucket.
+
+ :param bucket_name: name of the bucket
+ :param key_name: name of the key to add
+ :param body: optional content of the key to add
+ """
+ cmd = ["aws", "s3api", "put-object", "--bucket", bucket_name, "--key", key_name]
Review comment:
True, but my thought is that we should not system-test our prod code with prod code? I mean theoretically if we change our code to use a different lib or so we also might need to change our test 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #8877: Add S3ToRedshift example dag and system test
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #8877:
URL: https://github.com/apache/airflow/pull/8877#discussion_r425906307
##########
File path: tests/providers/amazon/aws/operators/test_s3_to_redshift_system.py
##########
@@ -0,0 +1,62 @@
+#
+# 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 tempfile import NamedTemporaryFile
+
+import pytest
+
+from airflow.providers.amazon.aws.example_dags.example_s3_to_redshift import (
+ REDSHIFT_SCHEMA, REDSHIFT_TABLE, S3_BUCKET, S3_KEY,
+)
+from tests.test_utils.amazon_system_helpers import (
+ AWS_DAG_FOLDER, AmazonSystemTest, provide_aws_context, provide_aws_s3_bucket, provide_aws_s3_key,
+)
+from tests.test_utils.postgres_system_helpers import provide_postgres_schema, provide_postgres_table
+
+
+@pytest.fixture
+def provide_s3_bucket_with_key():
+ with provide_aws_s3_bucket(name=S3_BUCKET, region='eu-central-1') as s3_bucket:
Review comment:
I needed to add the `region` parameter so the s3 bucket will be created in that region. Because otherwise it would be in `us` and my existing cluster is in `eu`. The `COPY` command does not work across regions. :/
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #8877: Add S3ToRedshift example dag and system test
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #8877:
URL: https://github.com/apache/airflow/pull/8877#discussion_r436064245
##########
File path: airflow/providers/amazon/aws/example_dags/example_s3_to_redshift.py
##########
@@ -41,13 +42,26 @@
schedule_interval=None,
tags=['example']
) as dag:
+ preparation__task_load_sample_data_to_s3 = PythonOperator(
+ python_callable=lambda: S3Hook().load_string("0,Airflow", f'{S3_KEY}/{REDSHIFT_TABLE}', S3_BUCKET),
+ task_id='preparation__load_sample_data_to_s3'
+ )
+ preparation__task_create_table = PostgresOperator(
+ sql=f'CREATE TABLE {REDSHIFT_TABLE}(Id int, Name varchar)',
+ postgres_conn_id='redshift_default',
+ task_id='preparation__create_table'
+ )
Review comment:
WDYT of having some `preparation` tasks? Not everything I can prepare with terraform.
##########
File path: scripts/ci/docker-compose/local.yml
##########
@@ -62,6 +62,7 @@ services:
- HOST_USER_ID
- HOST_GROUP_ID
- HOST_HOME=${HOME}
+ - HOST_AIRFLOW_SOURCES
Review comment:
and here.
##########
File path: Dockerfile.ci
##########
@@ -308,26 +308,32 @@ RUN if [[ -n "${ADDITIONAL_PYTHON_DEPS}" ]]; then \
pip install ${ADDITIONAL_PYTHON_DEPS}; \
fi
+# Additional binaries to load if needed
RUN \
AWSCLI_IMAGE="amazon/aws-cli:latest" && \
AZURECLI_IMAGE="mcr.microsoft.com/azure-cli:latest" && \
GCLOUD_IMAGE="gcr.io/google.com/cloudsdktool/cloud-sdk:latest" && \
- echo "docker run --rm -it -v \${HOST_HOME}/.aws:/root/.aws ${AWSCLI_IMAGE} \"\$@\"" \
- > /usr/bin/aws && \
+ TERRAFORM_IMAGE="hashicorp/terraform:latest" && \
+ echo "#!/bin/bash" | tee -a /usr/bin/aws /usr/bin/az /usr/bin/bq /usr/bin/gcloud /usr/bin/gsutil /usr/bin/terraform && \
+ echo "docker run --rm -it -v \${HOST_HOME}/.aws:/root/.aws -v \${HOST_AIRFLOW_SOURCES}:/opt/airflow ${AWSCLI_IMAGE} \"\$@\"" \
+ >> /usr/bin/aws && \
echo "docker pull ${AWSCLI_IMAGE}" > /usr/bin/aws-update && \
- echo "docker run --rm -it -v \${HOST_HOME}/.azure:/root/.azure ${AZURECLI_IMAGE} \"\$@\"" \
- > /usr/bin/az && \
+ echo "docker run --rm -it -v \${HOST_HOME}/.azure:/root/.azure -v \${HOST_AIRFLOW_SOURCES}:/opt/airflow ${AZURECLI_IMAGE} \"\$@\"" \
+ >> /usr/bin/az && \
echo "docker pull ${AZURECLI_IMAGE}" > /usr/bin/az-update && \
- echo "docker run --rm -it -v \${HOST_HOME}/.config:/root/.config ${GCLOUD_IMAGE} bq \"\$@\"" \
- > /usr/bin/bq && \
+ echo "docker run --rm -it -v \${HOST_HOME}/.config:/root/.config -v \${HOST_AIRFLOW_SOURCES}:/opt/airflow ${GCLOUD_IMAGE} bq \"\$@\"" \
+ >> /usr/bin/bq && \
echo "docker pull ${GCLOUD_IMAGE}" > /usr/bin/bq-update && \
- echo "docker run --rm -it -v \${HOST_HOME}/.config:/root/.config ${GCLOUD_IMAGE} gcloud \"\$@\"" \
- > /usr/bin/gcloud && \
+ echo "docker run --rm -it -v \${HOST_HOME}/.config:/root/.config -v \${HOST_AIRFLOW_SOURCES}:/opt/airflow ${GCLOUD_IMAGE} gcloud \"\$@\"" \
+ >> /usr/bin/gcloud && \
echo "docker pull ${GCLOUD_IMAGE}" > /usr/bin/gcloud-update && \
- echo "docker run --rm -it -v \${HOST_HOME}/.config:/root/.config ${GCLOUD_IMAGE} gsutil \"\$@\"" \
- > /usr/bin/gsutil && \
+ echo "docker run --rm -it -v \${HOST_HOME}/.config:/root/.config -v \${HOST_AIRFLOW_SOURCES}:/opt/airflow ${GCLOUD_IMAGE} gsutil \"\$@\"" \
+ >> /usr/bin/gsutil && \
echo "docker pull ${GCLOUD_IMAGE}" > /usr/bin/gsutil-update && \
- chmod a+x /usr/bin/aws /usr/bin/az /usr/bin/bq /usr/bin/gcloud /usr/bin/gsutil
+ echo "docker run --rm -it -v \${HOST_HOME}/.aws:/root/.aws -v \${HOST_HOME}/.azure:/root/.azure -v \${HOST_HOME}/.config:/root/.config -v \${HOST_AIRFLOW_SOURCES}:/opt/airflow -w /opt/airflow --env-file <(env | grep TF) ${TERRAFORM_IMAGE} \"\$@\"" \
+ >> /usr/bin/terraform && \
Review comment:
Here I added all config dirs from aws, azure and google so terraform can use those to authenticate.
I also added `-v \${HOST_AIRFLOW_SOURCES}:/opt/airflow` to all of them so they can use airflow sources. For example terraform needs to access its `*.tf` and `*.tfvars` files.
`--env-file <(env | grep TF)` <- this is an interesting one. I did this to share all `TF*` env variables. So you can define those in `variables.env` from breeze and we do not need another one. Unfortunatly we can't just do `/files/airflow-breeze-config/variables.env` because this env file also has bash syntax in it - not only env var declarations.
##########
File path: scripts/ci/docker-compose/local-prod.yml
##########
@@ -40,4 +40,5 @@ services:
- HOST_USER_ID
- HOST_GROUP_ID
- HOST_HOME=${HOME}
+ - HOST_AIRFLOW_SOURCES
Review comment:
@potiuk I added `HOST_AIRFLOW_SOURCES` here and set it in my `variables.env` file. WDYT?
##########
File path: Dockerfile.ci
##########
@@ -308,26 +308,32 @@ RUN if [[ -n "${ADDITIONAL_PYTHON_DEPS}" ]]; then \
pip install ${ADDITIONAL_PYTHON_DEPS}; \
fi
+# Additional binaries to load if needed
RUN \
AWSCLI_IMAGE="amazon/aws-cli:latest" && \
AZURECLI_IMAGE="mcr.microsoft.com/azure-cli:latest" && \
GCLOUD_IMAGE="gcr.io/google.com/cloudsdktool/cloud-sdk:latest" && \
- echo "docker run --rm -it -v \${HOST_HOME}/.aws:/root/.aws ${AWSCLI_IMAGE} \"\$@\"" \
- > /usr/bin/aws && \
+ TERRAFORM_IMAGE="hashicorp/terraform:latest" && \
+ echo "#!/bin/bash" | tee -a /usr/bin/aws /usr/bin/az /usr/bin/bq /usr/bin/gcloud /usr/bin/gsutil /usr/bin/terraform && \
Review comment:
I added this line so python can run them properly - all bin's need to have a `#!/bin/bash` header.
##########
File path: tests/test_utils/terraform.py
##########
@@ -0,0 +1,36 @@
+# 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 tests.test_utils.system_tests_class import SystemTest
+
+
+class Terraform(SystemTest):
+ TERRAFORM_DIR: str
+
+ def setUp(self) -> None:
+ self.execute_cmd(["terraform", "init", "-input=false", self.TERRAFORM_DIR])
+ self.execute_cmd(["terraform", "plan", "-input=false", self.TERRAFORM_DIR])
+ self.execute_cmd(["terraform", "apply", "-input=false", "-auto-approve", self.TERRAFORM_DIR])
+
+ def get_tf_output(self, name):
+ output = self.check_output(["terraform", "output", name]).decode('utf-8').replace("\r\n", "")
+ self.log.info(output)
+ return output
+
+ def tearDown(self) -> None:
+ self.execute_cmd(["terraform", "plan", "-destroy", "-input=false", self.TERRAFORM_DIR])
+ self.execute_cmd(["terraform", "destroy", "-input=false", "-auto-approve", self.TERRAFORM_DIR])
Review comment:
I am not quite happy about it. This can probably be structured better.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on a change in pull request #8877: Add S3ToRedshift example dag and system test
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #8877:
URL: https://github.com/apache/airflow/pull/8877#discussion_r437211600
##########
File path: airflow/providers/amazon/aws/example_dags/example_s3_to_redshift.py
##########
@@ -41,13 +42,26 @@
schedule_interval=None,
tags=['example']
) as dag:
+ preparation__task_load_sample_data_to_s3 = PythonOperator(
+ python_callable=lambda: S3Hook().load_string("0,Airflow", f'{S3_KEY}/{REDSHIFT_TABLE}', S3_BUCKET),
+ task_id='preparation__load_sample_data_to_s3'
+ )
+ preparation__task_create_table = PostgresOperator(
+ sql=f'CREATE TABLE {REDSHIFT_TABLE}(Id int, Name varchar)',
+ postgres_conn_id='redshift_default',
+ task_id='preparation__create_table'
+ )
Review comment:
That's what we usually tried to do in GCP, make example that is self-sufficient so I'm ok with this approach. However, this requires few things:
- operator must be idempotent ("setup" steps in example must work always)
- consistency between what is created in example and what is deleted in test tear down
##########
File path: tests/providers/amazon/aws/operators/test_s3_to_redshift_system.py
##########
@@ -0,0 +1,44 @@
+#
+# 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.
+import os
+
+import pytest
+
+from airflow.models import Connection
+from airflow.utils import db
+from tests.test_utils import AIRFLOW_MAIN_FOLDER
+from tests.test_utils.amazon_system_helpers import AWS_DAG_FOLDER, AmazonSystemTest
+from tests.test_utils.terraform import Terraform
+
+
+@pytest.mark.backend("mysql", "postgres")
+class TestS3ToRedshiftExampleDags(AmazonSystemTest, Terraform):
+ TERRAFORM_DIR = os.path.join(
+ AIRFLOW_MAIN_FOLDER, "tests", "providers", "amazon", "aws", "infrastructure", "example_s3_to_redshift"
+ )
+
+ def setUp(self) -> None:
+ super().setUp()
+ host, port = self.get_tf_output("redshift_endpoint").split(':')
+ schema = self.get_tf_output("redshift_database_name"),
+ login = self.get_tf_output("redshift_master_username"),
+ password = self.get_tf_output("redshift_master_password")
+ db.merge_conn(Connection("redshift_default", "postgres", host, login, password, schema, port))
Review comment:
Is this connection deleted after test?
##########
File path: tests/test_utils/terraform.py
##########
@@ -0,0 +1,36 @@
+# 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 tests.test_utils.system_tests_class import SystemTest
+
+
+class Terraform(SystemTest):
+ TERRAFORM_DIR: str
+
+ def setUp(self) -> None:
+ self.execute_cmd(["terraform", "init", "-input=false", self.TERRAFORM_DIR])
+ self.execute_cmd(["terraform", "plan", "-input=false", self.TERRAFORM_DIR])
+ self.execute_cmd(["terraform", "apply", "-input=false", "-auto-approve", self.TERRAFORM_DIR])
+
+ def get_tf_output(self, name):
+ output = self.check_output(["terraform", "output", name]).decode('utf-8').replace("\r\n", "")
+ self.log.info(output)
+ return output
+
+ def tearDown(self) -> None:
+ self.execute_cmd(["terraform", "plan", "-destroy", "-input=false", self.TERRAFORM_DIR])
+ self.execute_cmd(["terraform", "destroy", "-input=false", "-auto-approve", self.TERRAFORM_DIR])
Review comment:
We can consider using/taking a look at https://github.com/beelit94/python-terraform
##########
File path: tests/test_utils/terraform.py
##########
@@ -0,0 +1,36 @@
+# 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 tests.test_utils.system_tests_class import SystemTest
+
+
+class Terraform(SystemTest):
+ TERRAFORM_DIR: str
+
+ def setUp(self) -> None:
+ self.execute_cmd(["terraform", "init", "-input=false", self.TERRAFORM_DIR])
+ self.execute_cmd(["terraform", "plan", "-input=false", self.TERRAFORM_DIR])
+ self.execute_cmd(["terraform", "apply", "-input=false", "-auto-approve", self.TERRAFORM_DIR])
+
+ def get_tf_output(self, name):
+ output = self.check_output(["terraform", "output", name]).decode('utf-8').replace("\r\n", "")
+ self.log.info(output)
+ return output
+
+ def tearDown(self) -> None:
+ self.execute_cmd(["terraform", "plan", "-destroy", "-input=false", self.TERRAFORM_DIR])
+ self.execute_cmd(["terraform", "destroy", "-input=false", "-auto-approve", self.TERRAFORM_DIR])
Review comment:
Then I would suggest to add `init`, `plan`, `apply` as methods/functions to limit the number of replicated code using `self.execute_cmd`. WDYT?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #8877: Add S3ToRedshift example dag and system test
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #8877:
URL: https://github.com/apache/airflow/pull/8877#discussion_r426175074
##########
File path: airflow/providers/amazon/aws/example_dags/example_s3_to_redshift.py
##########
@@ -0,0 +1,53 @@
+# 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 is an example dag for using `S3ToRedshiftTransfer` to copy a S3 key into a Redshift table.
+"""
+
+from os import getenv
+
+from airflow import DAG
+from airflow.providers.amazon.aws.operators.s3_to_redshift import S3ToRedshiftTransfer
+from airflow.utils.dates import days_ago
+
+# [START howto_operator_s3_to_redshift_env_variables]
+S3_BUCKET = getenv("S3_BUCKET", "bucket")
+S3_KEY = getenv("S3_KEY", "key")
Review comment:
Why are we relying on env vars, rather Airflow Variables+templates? `{{ var.s3_key}}`
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #8877: Add S3ToRedshift example dag and system test
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #8877:
URL: https://github.com/apache/airflow/pull/8877#discussion_r438204839
##########
File path: airflow/providers/amazon/aws/example_dags/example_s3_to_redshift.py
##########
@@ -0,0 +1,53 @@
+# 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 is an example dag for using `S3ToRedshiftTransfer` to copy a S3 key into a Redshift table.
+"""
+
+from os import getenv
+
+from airflow import DAG
+from airflow.providers.amazon.aws.operators.s3_to_redshift import S3ToRedshiftTransfer
+from airflow.utils.dates import days_ago
+
+# [START howto_operator_s3_to_redshift_env_variables]
+S3_BUCKET = getenv("S3_BUCKET", "bucket")
+S3_KEY = getenv("S3_KEY", "key")
Review comment:
I think this is fine and when we automate it, we should define files where we define ENV variables.
Using Airflow Variables in example DAGs is a very bad idea. People use it as - well - examples - and using airflow Variables in DAGs in this way is very bad from Database load point of view. Until we avoid parsing the DAGs by all entities frequently we should not use variables in this way (we even have best practice about it) - basically every time the DAG is parsed, DB communication occurs and the extra connection is made - that is quite bad for scalability and not obvious at all, so if people take that literally as examples (which they did in the past) they will implement bad practices.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #8877: Add S3ToRedshift example dag and system test
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #8877:
URL: https://github.com/apache/airflow/pull/8877#discussion_r426175472
##########
File path: tests/test_utils/postgres_system_helpers.py
##########
@@ -0,0 +1,97 @@
+# 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 contextlib import contextmanager
+from typing import List
+
+import pytest
+
+from tests.test_utils.system_tests_class import SystemTest
+from tests.utils.logging_command_executor import get_executor
+
+
+@contextmanager
+def provide_postgres_schema(name):
+ PostgresSystemTest.create_schema(name)
+ yield name
+ PostgresSystemTest.delete_schema(name)
+
+
+@contextmanager
+def provide_postgres_table(schema_name, table_name, columns=None):
+ PostgresSystemTest.create_table(schema_name, table_name, columns)
+ yield
+ PostgresSystemTest.delete_table(schema_name, table_name)
+
+
+@pytest.mark.system("postgres")
+class PostgresSystemTest(SystemTest):
+
+ @classmethod
+ def execute_with_ctx(cls, cmd: str):
+ """
+ Executes command with postgres (psql cli) context.
+ """
+ executor = get_executor()
+ from os import getenv
+ connection_string = getenv("AIRFLOW_CONN_POSTGRES_DEFAULT")
+ auth = ["psql", connection_string]
+ executor.execute_cmd(cmd=auth + ["-c", cmd])
Review comment:
Again, why are we shelling out rather than using DbAPiHook/psycopg2?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #8877: Add S3ToRedshift example dag and system test
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #8877:
URL: https://github.com/apache/airflow/pull/8877#discussion_r426180536
##########
File path: airflow/providers/amazon/aws/example_dags/example_s3_to_redshift.py
##########
@@ -0,0 +1,53 @@
+# 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 is an example dag for using `S3ToRedshiftTransfer` to copy a S3 key into a Redshift table.
+"""
+
+from os import getenv
+
+from airflow import DAG
+from airflow.providers.amazon.aws.operators.s3_to_redshift import S3ToRedshiftTransfer
+from airflow.utils.dates import days_ago
+
+# [START howto_operator_s3_to_redshift_env_variables]
+S3_BUCKET = getenv("S3_BUCKET", "bucket")
+S3_KEY = getenv("S3_KEY", "key")
Review comment:
We have this in all of our system tests. So that we can easily pass them from the host to breeze. See https://github.com/apache/airflow/blob/master/TESTING.rst#environment-for-system-tests
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #8877: Add S3ToRedshift example dag and system test
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #8877:
URL: https://github.com/apache/airflow/pull/8877#discussion_r426180306
##########
File path: tests/test_utils/amazon_system_helpers.py
##########
@@ -82,6 +89,31 @@ def delete_aws_s3_bucket(cls, name: str) -> None:
cmd = ["aws", "s3api", "delete-bucket", "--bucket", name]
cls.execute_with_ctx(cmd)
+ @classmethod
+ def create_aws_s3_key(cls, bucket_name: str, key_name: str, body: Optional[str] = None) -> None:
+ """
+ Puts an object into the aws bucket.
+
+ :param bucket_name: name of the bucket
+ :param key_name: name of the key to add
+ :param body: optional content of the key to add
+ """
+ cmd = ["aws", "s3api", "put-object", "--bucket", bucket_name, "--key", key_name]
Review comment:
Also some function we don't have in our hooks or operators. We then would need to mix hook, operator and boto3 calls. I think it would get confusing by having different abstraction layers.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #8877: Add S3ToRedshift example dag and system test
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #8877:
URL: https://github.com/apache/airflow/pull/8877#discussion_r426186170
##########
File path: tests/test_utils/postgres_system_helpers.py
##########
@@ -0,0 +1,97 @@
+# 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 contextlib import contextmanager
+from typing import List
+
+import pytest
+
+from tests.test_utils.system_tests_class import SystemTest
+from tests.utils.logging_command_executor import get_executor
+
+
+@contextmanager
+def provide_postgres_schema(name):
+ PostgresSystemTest.create_schema(name)
+ yield name
+ PostgresSystemTest.delete_schema(name)
+
+
+@contextmanager
+def provide_postgres_table(schema_name, table_name, columns=None):
Review comment:
```suggestion
def provide_postgres_table(schema_name, table_name, columns):
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #8877: [WIP] Add S3ToRedshift example dag and system test
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #8877:
URL: https://github.com/apache/airflow/pull/8877#discussion_r425901097
##########
File path: tests/test_utils/postgres_system_helpers.py
##########
@@ -0,0 +1,97 @@
+# 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 contextlib import contextmanager
+from typing import List
+
+import pytest
+
+from tests.test_utils.system_tests_class import SystemTest
+from tests.utils.logging_command_executor import get_executor
+
+
+@contextmanager
+def provide_postgres_schema(name):
+ PostgresSystemTest.create_schema(name)
+ yield name
+ PostgresSystemTest.delete_schema(name)
+
+
+@contextmanager
+def provide_postgres_table(schema_name, table_name, columns=None):
+ PostgresSystemTest.create_table(schema_name, table_name, columns)
+ yield
+ PostgresSystemTest.delete_table(schema_name, table_name)
+
+
+@pytest.mark.system("postgres")
+class PostgresSystemTest(SystemTest):
+
+ @classmethod
+ def execute_with_ctx(cls, cmd: str):
+ """
+ Executes command with postgres (psql cli) context.
+ """
+ executor = get_executor()
+ from os import getenv
+ connection_string = getenv("AIRFLOW_CONN_POSTGRES_DEFAULT")
+ auth = ["psql", connection_string]
+ executor.execute_cmd(cmd=auth + ["-c", cmd])
Review comment:
> from os import getenv
> connection_string = getenv("AIRFLOW_CONN_POSTGRES_DEFAULT")
Getting the connection through a env variable that needs to be defined :/ Is there a better way to do that?
> auth = ["psql", connection_string]
> executor.execute_cmd(cmd=auth + ["-c", cmd])
I don't like that part. I want to authenticate the context / the `psql` commands - that was the only way it worked. I also tried to authenticate the context through [Connection Service File](https://www.postgresql.org/docs/9.1/libpq-pgservice.html) and [Password File](https://www.postgresql.org/docs/9.1/libpq-pgpass.html) and mount them via `forward-credentials.yml` into the container.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #8877: [WIP] Add S3ToRedshift example dag and system test
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #8877:
URL: https://github.com/apache/airflow/pull/8877#discussion_r425901097
##########
File path: tests/test_utils/postgres_system_helpers.py
##########
@@ -0,0 +1,97 @@
+# 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 contextlib import contextmanager
+from typing import List
+
+import pytest
+
+from tests.test_utils.system_tests_class import SystemTest
+from tests.utils.logging_command_executor import get_executor
+
+
+@contextmanager
+def provide_postgres_schema(name):
+ PostgresSystemTest.create_schema(name)
+ yield name
+ PostgresSystemTest.delete_schema(name)
+
+
+@contextmanager
+def provide_postgres_table(schema_name, table_name, columns=None):
+ PostgresSystemTest.create_table(schema_name, table_name, columns)
+ yield
+ PostgresSystemTest.delete_table(schema_name, table_name)
+
+
+@pytest.mark.system("postgres")
+class PostgresSystemTest(SystemTest):
+
+ @classmethod
+ def execute_with_ctx(cls, cmd: str):
+ """
+ Executes command with postgres (psql cli) context.
+ """
+ executor = get_executor()
+ from os import getenv
+ connection_string = getenv("AIRFLOW_CONN_POSTGRES_DEFAULT")
+ auth = ["psql", connection_string]
+ executor.execute_cmd(cmd=auth + ["-c", cmd])
Review comment:
> from os import getenv
> connection_string = getenv("AIRFLOW_CONN_POSTGRES_DEFAULT")
Getting the connection through a env variable that needs to be defined :/ Is there a better way to do that?
> auth = ["psql", connection_string]
> executor.execute_cmd(cmd=auth + ["-c", cmd])
I don't like that part. I want to authenticate the context / the `psql` commands - that was only way it worked. I also tried to authenticate the context through [Connection Service File](https://www.postgresql.org/docs/9.1/libpq-pgservice.html) and [Password File](https://www.postgresql.org/docs/9.1/libpq-pgpass.html) and mount them via `forward-credentials.yml` into the container.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #8877: [WIP] Add S3ToRedshift example dag and system test
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #8877:
URL: https://github.com/apache/airflow/pull/8877#discussion_r425901097
##########
File path: tests/test_utils/postgres_system_helpers.py
##########
@@ -0,0 +1,97 @@
+# 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 contextlib import contextmanager
+from typing import List
+
+import pytest
+
+from tests.test_utils.system_tests_class import SystemTest
+from tests.utils.logging_command_executor import get_executor
+
+
+@contextmanager
+def provide_postgres_schema(name):
+ PostgresSystemTest.create_schema(name)
+ yield name
+ PostgresSystemTest.delete_schema(name)
+
+
+@contextmanager
+def provide_postgres_table(schema_name, table_name, columns=None):
+ PostgresSystemTest.create_table(schema_name, table_name, columns)
+ yield
+ PostgresSystemTest.delete_table(schema_name, table_name)
+
+
+@pytest.mark.system("postgres")
+class PostgresSystemTest(SystemTest):
+
+ @classmethod
+ def execute_with_ctx(cls, cmd: str):
+ """
+ Executes command with postgres (psql cli) context.
+ """
+ executor = get_executor()
+ from os import getenv
+ connection_string = getenv("AIRFLOW_CONN_POSTGRES_DEFAULT")
+ auth = ["psql", connection_string]
+ executor.execute_cmd(cmd=auth + ["-c", cmd])
Review comment:
> from os import getenv
> connection_string = getenv("AIRFLOW_CONN_POSTGRES_DEFAULT")
Getting the connection through a env variable that needs to be defined :/ Is there a better way to do that?
> auth = ["psql", connection_string]
> executor.execute_cmd(cmd=auth + ["-c", cmd])
I don't like that part. I want to authenticate the context / the `psql` commands - that was only thing that worked. I also tried to authenticate the context through [Connection Service File](https://www.postgresql.org/docs/9.1/libpq-pgservice.html) and [Password File](https://www.postgresql.org/docs/9.1/libpq-pgpass.html) and mount them via `forward-credentials.yml` into the container.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #8877: Add S3ToRedshift example dag and system test
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #8877:
URL: https://github.com/apache/airflow/pull/8877#discussion_r427186765
##########
File path: tests/test_utils/amazon_system_helpers.py
##########
@@ -82,6 +89,31 @@ def delete_aws_s3_bucket(cls, name: str) -> None:
cmd = ["aws", "s3api", "delete-bucket", "--bucket", name]
cls.execute_with_ctx(cmd)
+ @classmethod
+ def create_aws_s3_key(cls, bucket_name: str, key_name: str, body: Optional[str] = None) -> None:
+ """
+ Puts an object into the aws bucket.
+
+ :param bucket_name: name of the bucket
+ :param key_name: name of the key to add
+ :param body: optional content of the key to add
+ """
+ cmd = ["aws", "s3api", "put-object", "--bucket", bucket_name, "--key", key_name]
Review comment:
Of course, terraform would be better to create resources but since we have `awscli` installed anyway, I think it is not bad to start like this and switch to terraform once we have it.
##########
File path: tests/test_utils/amazon_system_helpers.py
##########
@@ -82,6 +89,31 @@ def delete_aws_s3_bucket(cls, name: str) -> None:
cmd = ["aws", "s3api", "delete-bucket", "--bucket", name]
cls.execute_with_ctx(cmd)
+ @classmethod
+ def create_aws_s3_key(cls, bucket_name: str, key_name: str, body: Optional[str] = None) -> None:
+ """
+ Puts an object into the aws bucket.
+
+ :param bucket_name: name of the bucket
+ :param key_name: name of the key to add
+ :param body: optional content of the key to add
+ """
+ cmd = ["aws", "s3api", "put-object", "--bucket", bucket_name, "--key", key_name]
Review comment:
@ashb Of course, terraform would be better to create resources but since we have `awscli` installed anyway, I think it is not bad to start like this and switch to terraform once we have 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #8877: Add S3ToRedshift example dag and system test
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #8877:
URL: https://github.com/apache/airflow/pull/8877#discussion_r436247961
##########
File path: airflow/providers/amazon/aws/example_dags/example_s3_to_redshift.py
##########
@@ -0,0 +1,53 @@
+# 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 is an example dag for using `S3ToRedshiftTransfer` to copy a S3 key into a Redshift table.
+"""
+
+from os import getenv
+
+from airflow import DAG
+from airflow.providers.amazon.aws.operators.s3_to_redshift import S3ToRedshiftTransfer
+from airflow.utils.dates import days_ago
+
+# [START howto_operator_s3_to_redshift_env_variables]
+S3_BUCKET = getenv("S3_BUCKET", "bucket")
+S3_KEY = getenv("S3_KEY", "key")
Review comment:
I think it makes sense to switch to Airflow Variables+templates like Ash said and set them via env variables: https://airflow.apache.org/docs/stable/concepts.html?highlight=xcom#storing-variables-in-environment-variables.
So we only need to prefix all env vars with `AIRFLOW_VAR`.
For example: set `AIRFLOW_VAR_S3_KEY=key` in `variables.env` and access them via `{{ var.s3_key}}` here. WDYT @potiuk @turbaszek ?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org