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/28 16:28:05 UTC
[GitHub] [airflow] JavierLopezT opened a new pull request #9054: Add MySqlToS3Operator
JavierLopezT opened a new pull request #9054:
URL: https://github.com/apache/airflow/pull/9054
Adding a new operator from transfering files from mysql to s3. This is a new PR for the same code of https://github.com/apache/airflow/pull/6670/files#, made for cleaning my git, which was giving me a lot of problems.
- [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/)"
- [x] 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).
----------------------------------------------------------------
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 #9054: Add MySqlToS3Operator
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #9054:
URL: https://github.com/apache/airflow/pull/9054#discussion_r432555050
##########
File path: airflow/providers/amazon/aws/operators/mysql_to_s3.py
##########
@@ -0,0 +1,126 @@
+#
+# 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 tempfile
+from typing import Optional, Union
+
+import numpy as np
+import pandas as pd
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.s3 import S3Hook
+from airflow.providers.mysql.hooks.mysql import MySqlHook
+from airflow.utils.decorators import apply_defaults
+
+
+class MySQLToS3Operator(BaseOperator):
+ """
+ Saves data from an specific MySQL query into a file in S3.
+
+ :param query: the sql query to be executed. If you want to execute a file, place the absolute path of it,
+ ending with .sql extension.
Review comment:
```suggestion
:param query: the sql query to be executed. If you want to execute a file, place the absolute path of it,
ending with .sql extension.
```
This should fix docs
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] feluelle commented on a change in pull request #9054: Add MySqlToS3Operator
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #9054:
URL: https://github.com/apache/airflow/pull/9054#discussion_r438028146
##########
File path: airflow/providers/amazon/aws/operators/mysql_to_s3.py
##########
@@ -0,0 +1,126 @@
+#
+# 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 tempfile
+from typing import Optional, Union
+
+import numpy as np
+import pandas as pd
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.s3 import S3Hook
+from airflow.providers.mysql.hooks.mysql import MySqlHook
+from airflow.utils.decorators import apply_defaults
+
+
+class MySQLToS3Operator(BaseOperator):
+ """
+ Saves data from an specific MySQL query into a file in S3.
+
+ :param query: the sql query to be executed. If you want to execute a file, place the absolute path of it,
+ ending with .sql extension.
+ :type query: str
+ :param s3_bucket: bucket where the data will be stored
+ :type s3_bucket: str
+ :param s3_key: desired key for the file. It includes the name of the file
+ :type s3_key: str
+ :param mysql_conn_id: reference to a specific mysql database
+ :type mysql_conn_id: str
+ :param aws_conn_id: reference to a specific S3 connection
+ :type aws_conn_id: str
+ :param verify: Whether or not to verify SSL certificates for S3 connection.
+ By default SSL certificates are verified.
+ You can provide the following values:
+
+ - ``False``: do not validate SSL certificates. SSL will still be used
+ (unless use_ssl is False), but SSL certificates will not be verified.
+ - ``path/to/cert/bundle.pem``: A filename of the CA cert bundle to uses.
+ You can specify this argument if you want to use a different
+ CA cert bundle than the one used by botocore.
+ :type verify: bool or str
+ :param pd_csv_kwargs: arguments to include in pd.to_csv (header, index, columns...)
+ :type pd_csv_kwargs: dict
+ :param index: whether to have the index or not in the dataframe
+ :type index: str
+ :param header: whether to include header or not into the S3 file
+ :type header: bool
+ """
+
+ template_fields = ('s3_key', 'query',)
+ template_ext = ('.sql',)
+
+ @apply_defaults
+ def __init__(
+ self,
+ query: str,
+ s3_bucket: str,
+ s3_key: str,
+ mysql_conn_id: str = 'mysql_default',
+ aws_conn_id: str = 'aws_default',
+ verify: Optional[Union[bool, str]] = None,
+ pd_csv_kwargs: Optional[dict] = None,
+ index: Optional[bool] = False,
+ header: Optional[bool] = False,
+ *args, **kwargs) -> None:
+ super().__init__(*args, **kwargs)
+ self.query = query
+ self.s3_bucket = s3_bucket
+ self.s3_key = s3_key
+ self.mysql_conn_id = mysql_conn_id
+ self.aws_conn_id = aws_conn_id
+ self.verify = verify
+
+ if not pd_csv_kwargs:
+ self.pd_csv_kwargs = {}
+ else:
+ self.pd_csv_kwargs = pd_csv_kwargs
+ if "index" not in self.pd_csv_kwargs:
+ self.pd_csv_kwargs["index"] = index
+ if "header" not in self.pd_csv_kwargs:
+ self.pd_csv_kwargs["header"] = header
+
+ def _fix_int_dtypes(self, df):
+ """
+ Mutate DataFrame to set dtypes for int columns containing NaN values."
+ """
+ for col in df:
+ if "float" in df[col].dtype.name and df[col].hasnans:
+ # inspect values to determine if dtype of non-null values is int or float
+ notna_series = df[col].dropna().values
+ if np.isclose(notna_series, notna_series.astype(int)).all():
+ # set to dtype that retains integers and supports NaNs
+ df[col] = np.where(df[col].isnull(), None, df[col]).astype(pd.Int64Dtype)
+
+ def execute(self, context):
+ mysql_hook = MySqlHook(mysql_conn_id=self.mysql_conn_id)
+ s3_conn = S3Hook(aws_conn_id=self.aws_conn_id, verify=self.verify)
+ data_df = mysql_hook.get_pandas_df(self.query)
+ self.log.info("Data from MySQL obtained")
+
+ self._fix_int_dtypes(data_df)
+ with tempfile.NamedTemporaryFile(mode='r+', suffix='.csv') as tmp_csv:
+ tmp_csv.file.write(data_df.to_csv(**self.pd_csv_kwargs))
+ tmp_csv.file.seek(0)
Review comment:
okay, but **Solution 1** should work. You should pass the file name not the object. Can you try this? `data_df.to_csv(tmp_csv.name, **self.pd_csv_kwargs)`
**EDIT:**
I looked up BytesIO again. BytesIO / StringIO does not create a file - only a buffer. So the "file" mentioned in your comment does not exist. cc @turbaszek
Also this explains it nicely in my opinion: https://stackoverflow.com/questions/42800250/difference-between-open-and-io-bytesio-in-binary-streams
----------------------------------------------------------------
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] JavierLopezT commented on a change in pull request #9054: Add MySqlToS3Operator
Posted by GitBox <gi...@apache.org>.
JavierLopezT commented on a change in pull request #9054:
URL: https://github.com/apache/airflow/pull/9054#discussion_r438827589
##########
File path: airflow/providers/amazon/aws/operators/mysql_to_s3.py
##########
@@ -0,0 +1,126 @@
+#
+# 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 tempfile
+from typing import Optional, Union
+
+import numpy as np
+import pandas as pd
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.s3 import S3Hook
+from airflow.providers.mysql.hooks.mysql import MySqlHook
+from airflow.utils.decorators import apply_defaults
+
+
+class MySQLToS3Operator(BaseOperator):
+ """
+ Saves data from an specific MySQL query into a file in S3.
+
+ :param query: the sql query to be executed. If you want to execute a file, place the absolute path of it,
+ ending with .sql extension.
+ :type query: str
+ :param s3_bucket: bucket where the data will be stored
+ :type s3_bucket: str
+ :param s3_key: desired key for the file. It includes the name of the file
+ :type s3_key: str
+ :param mysql_conn_id: reference to a specific mysql database
+ :type mysql_conn_id: str
+ :param aws_conn_id: reference to a specific S3 connection
+ :type aws_conn_id: str
+ :param verify: Whether or not to verify SSL certificates for S3 connection.
+ By default SSL certificates are verified.
+ You can provide the following values:
+
+ - ``False``: do not validate SSL certificates. SSL will still be used
+ (unless use_ssl is False), but SSL certificates will not be verified.
+ - ``path/to/cert/bundle.pem``: A filename of the CA cert bundle to uses.
+ You can specify this argument if you want to use a different
+ CA cert bundle than the one used by botocore.
+ :type verify: bool or str
+ :param pd_csv_kwargs: arguments to include in pd.to_csv (header, index, columns...)
+ :type pd_csv_kwargs: dict
+ :param index: whether to have the index or not in the dataframe
+ :type index: str
+ :param header: whether to include header or not into the S3 file
+ :type header: bool
+ """
+
+ template_fields = ('s3_key', 'query',)
+ template_ext = ('.sql',)
+
+ @apply_defaults
+ def __init__(
+ self,
+ query: str,
+ s3_bucket: str,
+ s3_key: str,
+ mysql_conn_id: str = 'mysql_default',
+ aws_conn_id: str = 'aws_default',
+ verify: Optional[Union[bool, str]] = None,
+ pd_csv_kwargs: Optional[dict] = None,
+ index: Optional[bool] = False,
+ header: Optional[bool] = False,
+ *args, **kwargs) -> None:
+ super().__init__(*args, **kwargs)
+ self.query = query
+ self.s3_bucket = s3_bucket
+ self.s3_key = s3_key
+ self.mysql_conn_id = mysql_conn_id
+ self.aws_conn_id = aws_conn_id
+ self.verify = verify
+
+ if not pd_csv_kwargs:
+ self.pd_csv_kwargs = {}
+ else:
+ self.pd_csv_kwargs = pd_csv_kwargs
+ if "index" not in self.pd_csv_kwargs:
+ self.pd_csv_kwargs["index"] = index
+ if "header" not in self.pd_csv_kwargs:
+ self.pd_csv_kwargs["header"] = header
+
+ def _fix_int_dtypes(self, df):
+ """
+ Mutate DataFrame to set dtypes for int columns containing NaN values."
+ """
+ for col in df:
+ if "float" in df[col].dtype.name and df[col].hasnans:
+ # inspect values to determine if dtype of non-null values is int or float
+ notna_series = df[col].dropna().values
+ if np.isclose(notna_series, notna_series.astype(int)).all():
+ # set to dtype that retains integers and supports NaNs
+ df[col] = np.where(df[col].isnull(), None, df[col]).astype(pd.Int64Dtype)
+
+ def execute(self, context):
+ mysql_hook = MySqlHook(mysql_conn_id=self.mysql_conn_id)
+ s3_conn = S3Hook(aws_conn_id=self.aws_conn_id, verify=self.verify)
+ data_df = mysql_hook.get_pandas_df(self.query)
+ self.log.info("Data from MySQL obtained")
+
+ self._fix_int_dtypes(data_df)
+ with tempfile.NamedTemporaryFile(mode='r+', suffix='.csv') as tmp_csv:
+ tmp_csv.file.write(data_df.to_csv(**self.pd_csv_kwargs))
+ tmp_csv.file.seek(0)
Review comment:
I made the changed and it worked. Thank you very much. Is there anything else I should to have your approval @feluelle ?
----------------------------------------------------------------
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 #9054: Add MySqlToS3Operator
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #9054:
URL: https://github.com/apache/airflow/pull/9054#discussion_r437967089
##########
File path: airflow/providers/amazon/aws/operators/mysql_to_s3.py
##########
@@ -0,0 +1,126 @@
+#
+# 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 tempfile
+from typing import Optional, Union
+
+import numpy as np
+import pandas as pd
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.s3 import S3Hook
+from airflow.providers.mysql.hooks.mysql import MySqlHook
+from airflow.utils.decorators import apply_defaults
+
+
+class MySQLToS3Operator(BaseOperator):
+ """
+ Saves data from an specific MySQL query into a file in S3.
+
+ :param query: the sql query to be executed. If you want to execute a file, place the absolute path of it,
+ ending with .sql extension.
+ :type query: str
+ :param s3_bucket: bucket where the data will be stored
+ :type s3_bucket: str
+ :param s3_key: desired key for the file. It includes the name of the file
+ :type s3_key: str
+ :param mysql_conn_id: reference to a specific mysql database
+ :type mysql_conn_id: str
+ :param aws_conn_id: reference to a specific S3 connection
+ :type aws_conn_id: str
+ :param verify: Whether or not to verify SSL certificates for S3 connection.
+ By default SSL certificates are verified.
+ You can provide the following values:
+
+ - ``False``: do not validate SSL certificates. SSL will still be used
+ (unless use_ssl is False), but SSL certificates will not be verified.
+ - ``path/to/cert/bundle.pem``: A filename of the CA cert bundle to uses.
+ You can specify this argument if you want to use a different
+ CA cert bundle than the one used by botocore.
+ :type verify: bool or str
+ :param pd_csv_kwargs: arguments to include in pd.to_csv (header, index, columns...)
+ :type pd_csv_kwargs: dict
+ :param index: whether to have the index or not in the dataframe
+ :type index: str
+ :param header: whether to include header or not into the S3 file
+ :type header: bool
+ """
+
+ template_fields = ('s3_key', 'query',)
+ template_ext = ('.sql',)
+
+ @apply_defaults
+ def __init__(
+ self,
+ query: str,
+ s3_bucket: str,
+ s3_key: str,
+ mysql_conn_id: str = 'mysql_default',
+ aws_conn_id: str = 'aws_default',
+ verify: Optional[Union[bool, str]] = None,
+ pd_csv_kwargs: Optional[dict] = None,
+ index: Optional[bool] = False,
+ header: Optional[bool] = False,
+ *args, **kwargs) -> None:
+ super().__init__(*args, **kwargs)
+ self.query = query
+ self.s3_bucket = s3_bucket
+ self.s3_key = s3_key
+ self.mysql_conn_id = mysql_conn_id
+ self.aws_conn_id = aws_conn_id
+ self.verify = verify
+
+ if not pd_csv_kwargs:
+ self.pd_csv_kwargs = {}
+ else:
+ self.pd_csv_kwargs = pd_csv_kwargs
+ if "index" not in self.pd_csv_kwargs:
+ self.pd_csv_kwargs["index"] = index
+ if "header" not in self.pd_csv_kwargs:
+ self.pd_csv_kwargs["header"] = header
+
+ def _fix_int_dtypes(self, df):
+ """
+ Mutate DataFrame to set dtypes for int columns containing NaN values."
+ """
+ for col in df:
+ if "float" in df[col].dtype.name and df[col].hasnans:
+ # inspect values to determine if dtype of non-null values is int or float
+ notna_series = df[col].dropna().values
+ if np.isclose(notna_series, notna_series.astype(int)).all():
+ # set to dtype that retains integers and supports NaNs
+ df[col] = np.where(df[col].isnull(), None, df[col]).astype(pd.Int64Dtype)
+
+ def execute(self, context):
+ mysql_hook = MySqlHook(mysql_conn_id=self.mysql_conn_id)
+ s3_conn = S3Hook(aws_conn_id=self.aws_conn_id, verify=self.verify)
+ data_df = mysql_hook.get_pandas_df(self.query)
+ self.log.info("Data from MySQL obtained")
+
+ self._fix_int_dtypes(data_df)
+ with tempfile.NamedTemporaryFile(mode='r+', suffix='.csv') as tmp_csv:
+ tmp_csv.file.write(data_df.to_csv(**self.pd_csv_kwargs))
+ tmp_csv.file.seek(0)
Review comment:
What happens if you pass a file to the current code (`tmp_csv.file.write(data_df.to_csv(**self.pd_csv_kwargs))`) through `pd_csv_kwargs`?
**Suggestion 1:**
Can't we do it like this i.e. write to the file through `to_csv` ?
```suggestion
with tempfile.NamedTemporaryFile(mode='r+', suffix='.csv') as tmp_csv:
data_df.to_csv(tmp_csv.name, **self.pd_csv_kwargs)
```
And validate `pd_csv_kwargs` to not contain `path_or_buf` because you create the file.
```
if "path_or_buf" in self.pd_csv_kwargs:
raise
```
**Suggestion 2:**
Write to s3 through `load_string` and pass the result of `data_df.to_csv(**self.pd_csv_kwargs)`. This way you do not need to create a temp file here. The file will be created by s3 hook.
But you should also validate `pd_csv_kwargs` to not contain `path_or_buf`.
##########
File path: airflow/providers/amazon/aws/operators/mysql_to_s3.py
##########
@@ -0,0 +1,126 @@
+#
+# 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 tempfile
+from typing import Optional, Union
+
+import numpy as np
+import pandas as pd
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.s3 import S3Hook
+from airflow.providers.mysql.hooks.mysql import MySqlHook
+from airflow.utils.decorators import apply_defaults
+
+
+class MySQLToS3Operator(BaseOperator):
+ """
+ Saves data from an specific MySQL query into a file in S3.
+
+ :param query: the sql query to be executed. If you want to execute a file, place the absolute path of it,
+ ending with .sql extension.
+ :type query: str
+ :param s3_bucket: bucket where the data will be stored
+ :type s3_bucket: str
+ :param s3_key: desired key for the file. It includes the name of the file
+ :type s3_key: str
+ :param mysql_conn_id: reference to a specific mysql database
+ :type mysql_conn_id: str
+ :param aws_conn_id: reference to a specific S3 connection
+ :type aws_conn_id: str
+ :param verify: Whether or not to verify SSL certificates for S3 connection.
+ By default SSL certificates are verified.
+ You can provide the following values:
+
+ - ``False``: do not validate SSL certificates. SSL will still be used
+ (unless use_ssl is False), but SSL certificates will not be verified.
+ - ``path/to/cert/bundle.pem``: A filename of the CA cert bundle to uses.
+ You can specify this argument if you want to use a different
+ CA cert bundle than the one used by botocore.
+ :type verify: bool or str
+ :param pd_csv_kwargs: arguments to include in pd.to_csv (header, index, columns...)
+ :type pd_csv_kwargs: dict
+ :param index: whether to have the index or not in the dataframe
+ :type index: str
+ :param header: whether to include header or not into the S3 file
+ :type header: bool
+ """
+
+ template_fields = ('s3_key', 'query',)
+ template_ext = ('.sql',)
+
+ @apply_defaults
+ def __init__(
+ self,
+ query: str,
+ s3_bucket: str,
+ s3_key: str,
+ mysql_conn_id: str = 'mysql_default',
+ aws_conn_id: str = 'aws_default',
+ verify: Optional[Union[bool, str]] = None,
+ pd_csv_kwargs: Optional[dict] = None,
+ index: Optional[bool] = False,
+ header: Optional[bool] = False,
+ *args, **kwargs) -> None:
+ super().__init__(*args, **kwargs)
+ self.query = query
+ self.s3_bucket = s3_bucket
+ self.s3_key = s3_key
+ self.mysql_conn_id = mysql_conn_id
+ self.aws_conn_id = aws_conn_id
+ self.verify = verify
+
+ if not pd_csv_kwargs:
+ self.pd_csv_kwargs = {}
+ else:
+ self.pd_csv_kwargs = pd_csv_kwargs
Review comment:
```suggestion
self.pd_csv_kwargs = pd_csv_kwargs or {}
```
----------------------------------------------------------------
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] JavierLopezT commented on a change in pull request #9054: Add MySqlToS3Operator
Posted by GitBox <gi...@apache.org>.
JavierLopezT commented on a change in pull request #9054:
URL: https://github.com/apache/airflow/pull/9054#discussion_r438827589
##########
File path: airflow/providers/amazon/aws/operators/mysql_to_s3.py
##########
@@ -0,0 +1,126 @@
+#
+# 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 tempfile
+from typing import Optional, Union
+
+import numpy as np
+import pandas as pd
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.s3 import S3Hook
+from airflow.providers.mysql.hooks.mysql import MySqlHook
+from airflow.utils.decorators import apply_defaults
+
+
+class MySQLToS3Operator(BaseOperator):
+ """
+ Saves data from an specific MySQL query into a file in S3.
+
+ :param query: the sql query to be executed. If you want to execute a file, place the absolute path of it,
+ ending with .sql extension.
+ :type query: str
+ :param s3_bucket: bucket where the data will be stored
+ :type s3_bucket: str
+ :param s3_key: desired key for the file. It includes the name of the file
+ :type s3_key: str
+ :param mysql_conn_id: reference to a specific mysql database
+ :type mysql_conn_id: str
+ :param aws_conn_id: reference to a specific S3 connection
+ :type aws_conn_id: str
+ :param verify: Whether or not to verify SSL certificates for S3 connection.
+ By default SSL certificates are verified.
+ You can provide the following values:
+
+ - ``False``: do not validate SSL certificates. SSL will still be used
+ (unless use_ssl is False), but SSL certificates will not be verified.
+ - ``path/to/cert/bundle.pem``: A filename of the CA cert bundle to uses.
+ You can specify this argument if you want to use a different
+ CA cert bundle than the one used by botocore.
+ :type verify: bool or str
+ :param pd_csv_kwargs: arguments to include in pd.to_csv (header, index, columns...)
+ :type pd_csv_kwargs: dict
+ :param index: whether to have the index or not in the dataframe
+ :type index: str
+ :param header: whether to include header or not into the S3 file
+ :type header: bool
+ """
+
+ template_fields = ('s3_key', 'query',)
+ template_ext = ('.sql',)
+
+ @apply_defaults
+ def __init__(
+ self,
+ query: str,
+ s3_bucket: str,
+ s3_key: str,
+ mysql_conn_id: str = 'mysql_default',
+ aws_conn_id: str = 'aws_default',
+ verify: Optional[Union[bool, str]] = None,
+ pd_csv_kwargs: Optional[dict] = None,
+ index: Optional[bool] = False,
+ header: Optional[bool] = False,
+ *args, **kwargs) -> None:
+ super().__init__(*args, **kwargs)
+ self.query = query
+ self.s3_bucket = s3_bucket
+ self.s3_key = s3_key
+ self.mysql_conn_id = mysql_conn_id
+ self.aws_conn_id = aws_conn_id
+ self.verify = verify
+
+ if not pd_csv_kwargs:
+ self.pd_csv_kwargs = {}
+ else:
+ self.pd_csv_kwargs = pd_csv_kwargs
+ if "index" not in self.pd_csv_kwargs:
+ self.pd_csv_kwargs["index"] = index
+ if "header" not in self.pd_csv_kwargs:
+ self.pd_csv_kwargs["header"] = header
+
+ def _fix_int_dtypes(self, df):
+ """
+ Mutate DataFrame to set dtypes for int columns containing NaN values."
+ """
+ for col in df:
+ if "float" in df[col].dtype.name and df[col].hasnans:
+ # inspect values to determine if dtype of non-null values is int or float
+ notna_series = df[col].dropna().values
+ if np.isclose(notna_series, notna_series.astype(int)).all():
+ # set to dtype that retains integers and supports NaNs
+ df[col] = np.where(df[col].isnull(), None, df[col]).astype(pd.Int64Dtype)
+
+ def execute(self, context):
+ mysql_hook = MySqlHook(mysql_conn_id=self.mysql_conn_id)
+ s3_conn = S3Hook(aws_conn_id=self.aws_conn_id, verify=self.verify)
+ data_df = mysql_hook.get_pandas_df(self.query)
+ self.log.info("Data from MySQL obtained")
+
+ self._fix_int_dtypes(data_df)
+ with tempfile.NamedTemporaryFile(mode='r+', suffix='.csv') as tmp_csv:
+ tmp_csv.file.write(data_df.to_csv(**self.pd_csv_kwargs))
+ tmp_csv.file.seek(0)
Review comment:
I made the changes and it worked. Thank you very much. Is there anything else I should do to have your approval @feluelle ?
----------------------------------------------------------------
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 #9054: Add MySqlToS3Operator
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #9054:
URL: https://github.com/apache/airflow/pull/9054#discussion_r438028146
##########
File path: airflow/providers/amazon/aws/operators/mysql_to_s3.py
##########
@@ -0,0 +1,126 @@
+#
+# 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 tempfile
+from typing import Optional, Union
+
+import numpy as np
+import pandas as pd
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.s3 import S3Hook
+from airflow.providers.mysql.hooks.mysql import MySqlHook
+from airflow.utils.decorators import apply_defaults
+
+
+class MySQLToS3Operator(BaseOperator):
+ """
+ Saves data from an specific MySQL query into a file in S3.
+
+ :param query: the sql query to be executed. If you want to execute a file, place the absolute path of it,
+ ending with .sql extension.
+ :type query: str
+ :param s3_bucket: bucket where the data will be stored
+ :type s3_bucket: str
+ :param s3_key: desired key for the file. It includes the name of the file
+ :type s3_key: str
+ :param mysql_conn_id: reference to a specific mysql database
+ :type mysql_conn_id: str
+ :param aws_conn_id: reference to a specific S3 connection
+ :type aws_conn_id: str
+ :param verify: Whether or not to verify SSL certificates for S3 connection.
+ By default SSL certificates are verified.
+ You can provide the following values:
+
+ - ``False``: do not validate SSL certificates. SSL will still be used
+ (unless use_ssl is False), but SSL certificates will not be verified.
+ - ``path/to/cert/bundle.pem``: A filename of the CA cert bundle to uses.
+ You can specify this argument if you want to use a different
+ CA cert bundle than the one used by botocore.
+ :type verify: bool or str
+ :param pd_csv_kwargs: arguments to include in pd.to_csv (header, index, columns...)
+ :type pd_csv_kwargs: dict
+ :param index: whether to have the index or not in the dataframe
+ :type index: str
+ :param header: whether to include header or not into the S3 file
+ :type header: bool
+ """
+
+ template_fields = ('s3_key', 'query',)
+ template_ext = ('.sql',)
+
+ @apply_defaults
+ def __init__(
+ self,
+ query: str,
+ s3_bucket: str,
+ s3_key: str,
+ mysql_conn_id: str = 'mysql_default',
+ aws_conn_id: str = 'aws_default',
+ verify: Optional[Union[bool, str]] = None,
+ pd_csv_kwargs: Optional[dict] = None,
+ index: Optional[bool] = False,
+ header: Optional[bool] = False,
+ *args, **kwargs) -> None:
+ super().__init__(*args, **kwargs)
+ self.query = query
+ self.s3_bucket = s3_bucket
+ self.s3_key = s3_key
+ self.mysql_conn_id = mysql_conn_id
+ self.aws_conn_id = aws_conn_id
+ self.verify = verify
+
+ if not pd_csv_kwargs:
+ self.pd_csv_kwargs = {}
+ else:
+ self.pd_csv_kwargs = pd_csv_kwargs
+ if "index" not in self.pd_csv_kwargs:
+ self.pd_csv_kwargs["index"] = index
+ if "header" not in self.pd_csv_kwargs:
+ self.pd_csv_kwargs["header"] = header
+
+ def _fix_int_dtypes(self, df):
+ """
+ Mutate DataFrame to set dtypes for int columns containing NaN values."
+ """
+ for col in df:
+ if "float" in df[col].dtype.name and df[col].hasnans:
+ # inspect values to determine if dtype of non-null values is int or float
+ notna_series = df[col].dropna().values
+ if np.isclose(notna_series, notna_series.astype(int)).all():
+ # set to dtype that retains integers and supports NaNs
+ df[col] = np.where(df[col].isnull(), None, df[col]).astype(pd.Int64Dtype)
+
+ def execute(self, context):
+ mysql_hook = MySqlHook(mysql_conn_id=self.mysql_conn_id)
+ s3_conn = S3Hook(aws_conn_id=self.aws_conn_id, verify=self.verify)
+ data_df = mysql_hook.get_pandas_df(self.query)
+ self.log.info("Data from MySQL obtained")
+
+ self._fix_int_dtypes(data_df)
+ with tempfile.NamedTemporaryFile(mode='r+', suffix='.csv') as tmp_csv:
+ tmp_csv.file.write(data_df.to_csv(**self.pd_csv_kwargs))
+ tmp_csv.file.seek(0)
Review comment:
okay, but **Solution 1** should work. You should pass the file name not the object. Can you try this? `data_df.to_csv(tmp_csv.name, **self.pd_csv_kwargs)`
**EDIT:**
I looked up BytesIO again. BytesIO / StringIO does not create a file - only a buffer. So the "file" mentioned in your comment does not exist. cc @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
[GitHub] [airflow] JavierLopezT commented on a change in pull request #9054: Add MySqlToS3Operator
Posted by GitBox <gi...@apache.org>.
JavierLopezT commented on a change in pull request #9054:
URL: https://github.com/apache/airflow/pull/9054#discussion_r438016768
##########
File path: airflow/providers/amazon/aws/operators/mysql_to_s3.py
##########
@@ -0,0 +1,126 @@
+#
+# 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 tempfile
+from typing import Optional, Union
+
+import numpy as np
+import pandas as pd
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.s3 import S3Hook
+from airflow.providers.mysql.hooks.mysql import MySqlHook
+from airflow.utils.decorators import apply_defaults
+
+
+class MySQLToS3Operator(BaseOperator):
+ """
+ Saves data from an specific MySQL query into a file in S3.
+
+ :param query: the sql query to be executed. If you want to execute a file, place the absolute path of it,
+ ending with .sql extension.
+ :type query: str
+ :param s3_bucket: bucket where the data will be stored
+ :type s3_bucket: str
+ :param s3_key: desired key for the file. It includes the name of the file
+ :type s3_key: str
+ :param mysql_conn_id: reference to a specific mysql database
+ :type mysql_conn_id: str
+ :param aws_conn_id: reference to a specific S3 connection
+ :type aws_conn_id: str
+ :param verify: Whether or not to verify SSL certificates for S3 connection.
+ By default SSL certificates are verified.
+ You can provide the following values:
+
+ - ``False``: do not validate SSL certificates. SSL will still be used
+ (unless use_ssl is False), but SSL certificates will not be verified.
+ - ``path/to/cert/bundle.pem``: A filename of the CA cert bundle to uses.
+ You can specify this argument if you want to use a different
+ CA cert bundle than the one used by botocore.
+ :type verify: bool or str
+ :param pd_csv_kwargs: arguments to include in pd.to_csv (header, index, columns...)
+ :type pd_csv_kwargs: dict
+ :param index: whether to have the index or not in the dataframe
+ :type index: str
+ :param header: whether to include header or not into the S3 file
+ :type header: bool
+ """
+
+ template_fields = ('s3_key', 'query',)
+ template_ext = ('.sql',)
+
+ @apply_defaults
+ def __init__(
+ self,
+ query: str,
+ s3_bucket: str,
+ s3_key: str,
+ mysql_conn_id: str = 'mysql_default',
+ aws_conn_id: str = 'aws_default',
+ verify: Optional[Union[bool, str]] = None,
+ pd_csv_kwargs: Optional[dict] = None,
+ index: Optional[bool] = False,
+ header: Optional[bool] = False,
+ *args, **kwargs) -> None:
+ super().__init__(*args, **kwargs)
+ self.query = query
+ self.s3_bucket = s3_bucket
+ self.s3_key = s3_key
+ self.mysql_conn_id = mysql_conn_id
+ self.aws_conn_id = aws_conn_id
+ self.verify = verify
+
+ if not pd_csv_kwargs:
+ self.pd_csv_kwargs = {}
+ else:
+ self.pd_csv_kwargs = pd_csv_kwargs
+ if "index" not in self.pd_csv_kwargs:
+ self.pd_csv_kwargs["index"] = index
+ if "header" not in self.pd_csv_kwargs:
+ self.pd_csv_kwargs["header"] = header
+
+ def _fix_int_dtypes(self, df):
+ """
+ Mutate DataFrame to set dtypes for int columns containing NaN values."
+ """
+ for col in df:
+ if "float" in df[col].dtype.name and df[col].hasnans:
+ # inspect values to determine if dtype of non-null values is int or float
+ notna_series = df[col].dropna().values
+ if np.isclose(notna_series, notna_series.astype(int)).all():
+ # set to dtype that retains integers and supports NaNs
+ df[col] = np.where(df[col].isnull(), None, df[col]).astype(pd.Int64Dtype)
+
+ def execute(self, context):
+ mysql_hook = MySqlHook(mysql_conn_id=self.mysql_conn_id)
+ s3_conn = S3Hook(aws_conn_id=self.aws_conn_id, verify=self.verify)
+ data_df = mysql_hook.get_pandas_df(self.query)
+ self.log.info("Data from MySQL obtained")
+
+ self._fix_int_dtypes(data_df)
+ with tempfile.NamedTemporaryFile(mode='r+', suffix='.csv') as tmp_csv:
+ tmp_csv.file.write(data_df.to_csv(**self.pd_csv_kwargs))
+ tmp_csv.file.seek(0)
Review comment:
Thank you very much for your suggestions.
Regarding validating pd_csv_kwargs, I will do it.
Regarding the other suggestion, I already tried to use just to_csv and I failed. Also, at the beginning, I was using StringIO (equivalent to BytesIO that s3 hook uses) and I was suggested to use the way I am using right now. Everything can be found in this discussion:
https://github.com/apache/airflow/pull/6670#discussion_r361645289
----------------------------------------------------------------
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 #9054: Add MySqlToS3Operator
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #9054:
URL: https://github.com/apache/airflow/pull/9054#discussion_r438028146
##########
File path: airflow/providers/amazon/aws/operators/mysql_to_s3.py
##########
@@ -0,0 +1,126 @@
+#
+# 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 tempfile
+from typing import Optional, Union
+
+import numpy as np
+import pandas as pd
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.s3 import S3Hook
+from airflow.providers.mysql.hooks.mysql import MySqlHook
+from airflow.utils.decorators import apply_defaults
+
+
+class MySQLToS3Operator(BaseOperator):
+ """
+ Saves data from an specific MySQL query into a file in S3.
+
+ :param query: the sql query to be executed. If you want to execute a file, place the absolute path of it,
+ ending with .sql extension.
+ :type query: str
+ :param s3_bucket: bucket where the data will be stored
+ :type s3_bucket: str
+ :param s3_key: desired key for the file. It includes the name of the file
+ :type s3_key: str
+ :param mysql_conn_id: reference to a specific mysql database
+ :type mysql_conn_id: str
+ :param aws_conn_id: reference to a specific S3 connection
+ :type aws_conn_id: str
+ :param verify: Whether or not to verify SSL certificates for S3 connection.
+ By default SSL certificates are verified.
+ You can provide the following values:
+
+ - ``False``: do not validate SSL certificates. SSL will still be used
+ (unless use_ssl is False), but SSL certificates will not be verified.
+ - ``path/to/cert/bundle.pem``: A filename of the CA cert bundle to uses.
+ You can specify this argument if you want to use a different
+ CA cert bundle than the one used by botocore.
+ :type verify: bool or str
+ :param pd_csv_kwargs: arguments to include in pd.to_csv (header, index, columns...)
+ :type pd_csv_kwargs: dict
+ :param index: whether to have the index or not in the dataframe
+ :type index: str
+ :param header: whether to include header or not into the S3 file
+ :type header: bool
+ """
+
+ template_fields = ('s3_key', 'query',)
+ template_ext = ('.sql',)
+
+ @apply_defaults
+ def __init__(
+ self,
+ query: str,
+ s3_bucket: str,
+ s3_key: str,
+ mysql_conn_id: str = 'mysql_default',
+ aws_conn_id: str = 'aws_default',
+ verify: Optional[Union[bool, str]] = None,
+ pd_csv_kwargs: Optional[dict] = None,
+ index: Optional[bool] = False,
+ header: Optional[bool] = False,
+ *args, **kwargs) -> None:
+ super().__init__(*args, **kwargs)
+ self.query = query
+ self.s3_bucket = s3_bucket
+ self.s3_key = s3_key
+ self.mysql_conn_id = mysql_conn_id
+ self.aws_conn_id = aws_conn_id
+ self.verify = verify
+
+ if not pd_csv_kwargs:
+ self.pd_csv_kwargs = {}
+ else:
+ self.pd_csv_kwargs = pd_csv_kwargs
+ if "index" not in self.pd_csv_kwargs:
+ self.pd_csv_kwargs["index"] = index
+ if "header" not in self.pd_csv_kwargs:
+ self.pd_csv_kwargs["header"] = header
+
+ def _fix_int_dtypes(self, df):
+ """
+ Mutate DataFrame to set dtypes for int columns containing NaN values."
+ """
+ for col in df:
+ if "float" in df[col].dtype.name and df[col].hasnans:
+ # inspect values to determine if dtype of non-null values is int or float
+ notna_series = df[col].dropna().values
+ if np.isclose(notna_series, notna_series.astype(int)).all():
+ # set to dtype that retains integers and supports NaNs
+ df[col] = np.where(df[col].isnull(), None, df[col]).astype(pd.Int64Dtype)
+
+ def execute(self, context):
+ mysql_hook = MySqlHook(mysql_conn_id=self.mysql_conn_id)
+ s3_conn = S3Hook(aws_conn_id=self.aws_conn_id, verify=self.verify)
+ data_df = mysql_hook.get_pandas_df(self.query)
+ self.log.info("Data from MySQL obtained")
+
+ self._fix_int_dtypes(data_df)
+ with tempfile.NamedTemporaryFile(mode='r+', suffix='.csv') as tmp_csv:
+ tmp_csv.file.write(data_df.to_csv(**self.pd_csv_kwargs))
+ tmp_csv.file.seek(0)
Review comment:
okay, but **Solution 1** should work. You should pass the file name not the object. Can you try this? `data_df.to_csv(tmp_csv.name, **self.pd_csv_kwargs)`
**EDIT:**
I looked up BytesIO again. BytesIO / StringIO does not create a file - only a buffer. So the "file" mentioned in your comment does not exist. cc @turbaszek
Also this explains it nicely in my opinion: https://stackoverflow.com/a/42800629
----------------------------------------------------------------
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 #9054: Add MySqlToS3Operator
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #9054:
URL: https://github.com/apache/airflow/pull/9054#discussion_r438028146
##########
File path: airflow/providers/amazon/aws/operators/mysql_to_s3.py
##########
@@ -0,0 +1,126 @@
+#
+# 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 tempfile
+from typing import Optional, Union
+
+import numpy as np
+import pandas as pd
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.s3 import S3Hook
+from airflow.providers.mysql.hooks.mysql import MySqlHook
+from airflow.utils.decorators import apply_defaults
+
+
+class MySQLToS3Operator(BaseOperator):
+ """
+ Saves data from an specific MySQL query into a file in S3.
+
+ :param query: the sql query to be executed. If you want to execute a file, place the absolute path of it,
+ ending with .sql extension.
+ :type query: str
+ :param s3_bucket: bucket where the data will be stored
+ :type s3_bucket: str
+ :param s3_key: desired key for the file. It includes the name of the file
+ :type s3_key: str
+ :param mysql_conn_id: reference to a specific mysql database
+ :type mysql_conn_id: str
+ :param aws_conn_id: reference to a specific S3 connection
+ :type aws_conn_id: str
+ :param verify: Whether or not to verify SSL certificates for S3 connection.
+ By default SSL certificates are verified.
+ You can provide the following values:
+
+ - ``False``: do not validate SSL certificates. SSL will still be used
+ (unless use_ssl is False), but SSL certificates will not be verified.
+ - ``path/to/cert/bundle.pem``: A filename of the CA cert bundle to uses.
+ You can specify this argument if you want to use a different
+ CA cert bundle than the one used by botocore.
+ :type verify: bool or str
+ :param pd_csv_kwargs: arguments to include in pd.to_csv (header, index, columns...)
+ :type pd_csv_kwargs: dict
+ :param index: whether to have the index or not in the dataframe
+ :type index: str
+ :param header: whether to include header or not into the S3 file
+ :type header: bool
+ """
+
+ template_fields = ('s3_key', 'query',)
+ template_ext = ('.sql',)
+
+ @apply_defaults
+ def __init__(
+ self,
+ query: str,
+ s3_bucket: str,
+ s3_key: str,
+ mysql_conn_id: str = 'mysql_default',
+ aws_conn_id: str = 'aws_default',
+ verify: Optional[Union[bool, str]] = None,
+ pd_csv_kwargs: Optional[dict] = None,
+ index: Optional[bool] = False,
+ header: Optional[bool] = False,
+ *args, **kwargs) -> None:
+ super().__init__(*args, **kwargs)
+ self.query = query
+ self.s3_bucket = s3_bucket
+ self.s3_key = s3_key
+ self.mysql_conn_id = mysql_conn_id
+ self.aws_conn_id = aws_conn_id
+ self.verify = verify
+
+ if not pd_csv_kwargs:
+ self.pd_csv_kwargs = {}
+ else:
+ self.pd_csv_kwargs = pd_csv_kwargs
+ if "index" not in self.pd_csv_kwargs:
+ self.pd_csv_kwargs["index"] = index
+ if "header" not in self.pd_csv_kwargs:
+ self.pd_csv_kwargs["header"] = header
+
+ def _fix_int_dtypes(self, df):
+ """
+ Mutate DataFrame to set dtypes for int columns containing NaN values."
+ """
+ for col in df:
+ if "float" in df[col].dtype.name and df[col].hasnans:
+ # inspect values to determine if dtype of non-null values is int or float
+ notna_series = df[col].dropna().values
+ if np.isclose(notna_series, notna_series.astype(int)).all():
+ # set to dtype that retains integers and supports NaNs
+ df[col] = np.where(df[col].isnull(), None, df[col]).astype(pd.Int64Dtype)
+
+ def execute(self, context):
+ mysql_hook = MySqlHook(mysql_conn_id=self.mysql_conn_id)
+ s3_conn = S3Hook(aws_conn_id=self.aws_conn_id, verify=self.verify)
+ data_df = mysql_hook.get_pandas_df(self.query)
+ self.log.info("Data from MySQL obtained")
+
+ self._fix_int_dtypes(data_df)
+ with tempfile.NamedTemporaryFile(mode='r+', suffix='.csv') as tmp_csv:
+ tmp_csv.file.write(data_df.to_csv(**self.pd_csv_kwargs))
+ tmp_csv.file.seek(0)
Review comment:
okay, but then **Solution 1** should work. You should pass the file name not the object. Can you try this? `data_df.to_csv(tmp_csv.name, **self.pd_csv_kwargs)`
##########
File path: airflow/providers/amazon/aws/operators/mysql_to_s3.py
##########
@@ -0,0 +1,126 @@
+#
+# 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 tempfile
+from typing import Optional, Union
+
+import numpy as np
+import pandas as pd
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.s3 import S3Hook
+from airflow.providers.mysql.hooks.mysql import MySqlHook
+from airflow.utils.decorators import apply_defaults
+
+
+class MySQLToS3Operator(BaseOperator):
+ """
+ Saves data from an specific MySQL query into a file in S3.
+
+ :param query: the sql query to be executed. If you want to execute a file, place the absolute path of it,
+ ending with .sql extension.
+ :type query: str
+ :param s3_bucket: bucket where the data will be stored
+ :type s3_bucket: str
+ :param s3_key: desired key for the file. It includes the name of the file
+ :type s3_key: str
+ :param mysql_conn_id: reference to a specific mysql database
+ :type mysql_conn_id: str
+ :param aws_conn_id: reference to a specific S3 connection
+ :type aws_conn_id: str
+ :param verify: Whether or not to verify SSL certificates for S3 connection.
+ By default SSL certificates are verified.
+ You can provide the following values:
+
+ - ``False``: do not validate SSL certificates. SSL will still be used
+ (unless use_ssl is False), but SSL certificates will not be verified.
+ - ``path/to/cert/bundle.pem``: A filename of the CA cert bundle to uses.
+ You can specify this argument if you want to use a different
+ CA cert bundle than the one used by botocore.
+ :type verify: bool or str
+ :param pd_csv_kwargs: arguments to include in pd.to_csv (header, index, columns...)
+ :type pd_csv_kwargs: dict
+ :param index: whether to have the index or not in the dataframe
+ :type index: str
+ :param header: whether to include header or not into the S3 file
+ :type header: bool
+ """
+
+ template_fields = ('s3_key', 'query',)
+ template_ext = ('.sql',)
+
+ @apply_defaults
+ def __init__(
+ self,
+ query: str,
+ s3_bucket: str,
+ s3_key: str,
+ mysql_conn_id: str = 'mysql_default',
+ aws_conn_id: str = 'aws_default',
+ verify: Optional[Union[bool, str]] = None,
+ pd_csv_kwargs: Optional[dict] = None,
+ index: Optional[bool] = False,
+ header: Optional[bool] = False,
+ *args, **kwargs) -> None:
+ super().__init__(*args, **kwargs)
+ self.query = query
+ self.s3_bucket = s3_bucket
+ self.s3_key = s3_key
+ self.mysql_conn_id = mysql_conn_id
+ self.aws_conn_id = aws_conn_id
+ self.verify = verify
+
+ if not pd_csv_kwargs:
+ self.pd_csv_kwargs = {}
+ else:
+ self.pd_csv_kwargs = pd_csv_kwargs
+ if "index" not in self.pd_csv_kwargs:
+ self.pd_csv_kwargs["index"] = index
+ if "header" not in self.pd_csv_kwargs:
+ self.pd_csv_kwargs["header"] = header
+
+ def _fix_int_dtypes(self, df):
+ """
+ Mutate DataFrame to set dtypes for int columns containing NaN values."
+ """
+ for col in df:
+ if "float" in df[col].dtype.name and df[col].hasnans:
+ # inspect values to determine if dtype of non-null values is int or float
+ notna_series = df[col].dropna().values
+ if np.isclose(notna_series, notna_series.astype(int)).all():
+ # set to dtype that retains integers and supports NaNs
+ df[col] = np.where(df[col].isnull(), None, df[col]).astype(pd.Int64Dtype)
+
+ def execute(self, context):
+ mysql_hook = MySqlHook(mysql_conn_id=self.mysql_conn_id)
+ s3_conn = S3Hook(aws_conn_id=self.aws_conn_id, verify=self.verify)
+ data_df = mysql_hook.get_pandas_df(self.query)
+ self.log.info("Data from MySQL obtained")
+
+ self._fix_int_dtypes(data_df)
+ with tempfile.NamedTemporaryFile(mode='r+', suffix='.csv') as tmp_csv:
+ tmp_csv.file.write(data_df.to_csv(**self.pd_csv_kwargs))
+ tmp_csv.file.seek(0)
Review comment:
okay, but **Solution 1** should work. You should pass the file name not the object. Can you try this? `data_df.to_csv(tmp_csv.name, **self.pd_csv_kwargs)`
----------------------------------------------------------------
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 merged pull request #9054: Add MySqlToS3Operator
Posted by GitBox <gi...@apache.org>.
turbaszek merged pull request #9054:
URL: https://github.com/apache/airflow/pull/9054
----------------------------------------------------------------
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] JavierLopezT commented on a change in pull request #9054: Add MySqlToS3Operator
Posted by GitBox <gi...@apache.org>.
JavierLopezT commented on a change in pull request #9054:
URL: https://github.com/apache/airflow/pull/9054#discussion_r438827589
##########
File path: airflow/providers/amazon/aws/operators/mysql_to_s3.py
##########
@@ -0,0 +1,126 @@
+#
+# 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 tempfile
+from typing import Optional, Union
+
+import numpy as np
+import pandas as pd
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.s3 import S3Hook
+from airflow.providers.mysql.hooks.mysql import MySqlHook
+from airflow.utils.decorators import apply_defaults
+
+
+class MySQLToS3Operator(BaseOperator):
+ """
+ Saves data from an specific MySQL query into a file in S3.
+
+ :param query: the sql query to be executed. If you want to execute a file, place the absolute path of it,
+ ending with .sql extension.
+ :type query: str
+ :param s3_bucket: bucket where the data will be stored
+ :type s3_bucket: str
+ :param s3_key: desired key for the file. It includes the name of the file
+ :type s3_key: str
+ :param mysql_conn_id: reference to a specific mysql database
+ :type mysql_conn_id: str
+ :param aws_conn_id: reference to a specific S3 connection
+ :type aws_conn_id: str
+ :param verify: Whether or not to verify SSL certificates for S3 connection.
+ By default SSL certificates are verified.
+ You can provide the following values:
+
+ - ``False``: do not validate SSL certificates. SSL will still be used
+ (unless use_ssl is False), but SSL certificates will not be verified.
+ - ``path/to/cert/bundle.pem``: A filename of the CA cert bundle to uses.
+ You can specify this argument if you want to use a different
+ CA cert bundle than the one used by botocore.
+ :type verify: bool or str
+ :param pd_csv_kwargs: arguments to include in pd.to_csv (header, index, columns...)
+ :type pd_csv_kwargs: dict
+ :param index: whether to have the index or not in the dataframe
+ :type index: str
+ :param header: whether to include header or not into the S3 file
+ :type header: bool
+ """
+
+ template_fields = ('s3_key', 'query',)
+ template_ext = ('.sql',)
+
+ @apply_defaults
+ def __init__(
+ self,
+ query: str,
+ s3_bucket: str,
+ s3_key: str,
+ mysql_conn_id: str = 'mysql_default',
+ aws_conn_id: str = 'aws_default',
+ verify: Optional[Union[bool, str]] = None,
+ pd_csv_kwargs: Optional[dict] = None,
+ index: Optional[bool] = False,
+ header: Optional[bool] = False,
+ *args, **kwargs) -> None:
+ super().__init__(*args, **kwargs)
+ self.query = query
+ self.s3_bucket = s3_bucket
+ self.s3_key = s3_key
+ self.mysql_conn_id = mysql_conn_id
+ self.aws_conn_id = aws_conn_id
+ self.verify = verify
+
+ if not pd_csv_kwargs:
+ self.pd_csv_kwargs = {}
+ else:
+ self.pd_csv_kwargs = pd_csv_kwargs
+ if "index" not in self.pd_csv_kwargs:
+ self.pd_csv_kwargs["index"] = index
+ if "header" not in self.pd_csv_kwargs:
+ self.pd_csv_kwargs["header"] = header
+
+ def _fix_int_dtypes(self, df):
+ """
+ Mutate DataFrame to set dtypes for int columns containing NaN values."
+ """
+ for col in df:
+ if "float" in df[col].dtype.name and df[col].hasnans:
+ # inspect values to determine if dtype of non-null values is int or float
+ notna_series = df[col].dropna().values
+ if np.isclose(notna_series, notna_series.astype(int)).all():
+ # set to dtype that retains integers and supports NaNs
+ df[col] = np.where(df[col].isnull(), None, df[col]).astype(pd.Int64Dtype)
+
+ def execute(self, context):
+ mysql_hook = MySqlHook(mysql_conn_id=self.mysql_conn_id)
+ s3_conn = S3Hook(aws_conn_id=self.aws_conn_id, verify=self.verify)
+ data_df = mysql_hook.get_pandas_df(self.query)
+ self.log.info("Data from MySQL obtained")
+
+ self._fix_int_dtypes(data_df)
+ with tempfile.NamedTemporaryFile(mode='r+', suffix='.csv') as tmp_csv:
+ tmp_csv.file.write(data_df.to_csv(**self.pd_csv_kwargs))
+ tmp_csv.file.seek(0)
Review comment:
I made the changes and it worked. Thank you very much. Is there anything else I should to have your approval @feluelle ?
----------------------------------------------------------------
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 pull request #9054: Add MySqlToS3Operator
Posted by GitBox <gi...@apache.org>.
turbaszek commented on pull request #9054:
URL: https://github.com/apache/airflow/pull/9054#issuecomment-636027958
Hi @JavierLopezT happy to help deliver this finally. Can you take a look at the static checks?
----------------------------------------------------------------
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] JavierLopezT commented on a change in pull request #9054: Add MySqlToS3Operator
Posted by GitBox <gi...@apache.org>.
JavierLopezT commented on a change in pull request #9054:
URL: https://github.com/apache/airflow/pull/9054#discussion_r438016768
##########
File path: airflow/providers/amazon/aws/operators/mysql_to_s3.py
##########
@@ -0,0 +1,126 @@
+#
+# 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 tempfile
+from typing import Optional, Union
+
+import numpy as np
+import pandas as pd
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.s3 import S3Hook
+from airflow.providers.mysql.hooks.mysql import MySqlHook
+from airflow.utils.decorators import apply_defaults
+
+
+class MySQLToS3Operator(BaseOperator):
+ """
+ Saves data from an specific MySQL query into a file in S3.
+
+ :param query: the sql query to be executed. If you want to execute a file, place the absolute path of it,
+ ending with .sql extension.
+ :type query: str
+ :param s3_bucket: bucket where the data will be stored
+ :type s3_bucket: str
+ :param s3_key: desired key for the file. It includes the name of the file
+ :type s3_key: str
+ :param mysql_conn_id: reference to a specific mysql database
+ :type mysql_conn_id: str
+ :param aws_conn_id: reference to a specific S3 connection
+ :type aws_conn_id: str
+ :param verify: Whether or not to verify SSL certificates for S3 connection.
+ By default SSL certificates are verified.
+ You can provide the following values:
+
+ - ``False``: do not validate SSL certificates. SSL will still be used
+ (unless use_ssl is False), but SSL certificates will not be verified.
+ - ``path/to/cert/bundle.pem``: A filename of the CA cert bundle to uses.
+ You can specify this argument if you want to use a different
+ CA cert bundle than the one used by botocore.
+ :type verify: bool or str
+ :param pd_csv_kwargs: arguments to include in pd.to_csv (header, index, columns...)
+ :type pd_csv_kwargs: dict
+ :param index: whether to have the index or not in the dataframe
+ :type index: str
+ :param header: whether to include header or not into the S3 file
+ :type header: bool
+ """
+
+ template_fields = ('s3_key', 'query',)
+ template_ext = ('.sql',)
+
+ @apply_defaults
+ def __init__(
+ self,
+ query: str,
+ s3_bucket: str,
+ s3_key: str,
+ mysql_conn_id: str = 'mysql_default',
+ aws_conn_id: str = 'aws_default',
+ verify: Optional[Union[bool, str]] = None,
+ pd_csv_kwargs: Optional[dict] = None,
+ index: Optional[bool] = False,
+ header: Optional[bool] = False,
+ *args, **kwargs) -> None:
+ super().__init__(*args, **kwargs)
+ self.query = query
+ self.s3_bucket = s3_bucket
+ self.s3_key = s3_key
+ self.mysql_conn_id = mysql_conn_id
+ self.aws_conn_id = aws_conn_id
+ self.verify = verify
+
+ if not pd_csv_kwargs:
+ self.pd_csv_kwargs = {}
+ else:
+ self.pd_csv_kwargs = pd_csv_kwargs
+ if "index" not in self.pd_csv_kwargs:
+ self.pd_csv_kwargs["index"] = index
+ if "header" not in self.pd_csv_kwargs:
+ self.pd_csv_kwargs["header"] = header
+
+ def _fix_int_dtypes(self, df):
+ """
+ Mutate DataFrame to set dtypes for int columns containing NaN values."
+ """
+ for col in df:
+ if "float" in df[col].dtype.name and df[col].hasnans:
+ # inspect values to determine if dtype of non-null values is int or float
+ notna_series = df[col].dropna().values
+ if np.isclose(notna_series, notna_series.astype(int)).all():
+ # set to dtype that retains integers and supports NaNs
+ df[col] = np.where(df[col].isnull(), None, df[col]).astype(pd.Int64Dtype)
+
+ def execute(self, context):
+ mysql_hook = MySqlHook(mysql_conn_id=self.mysql_conn_id)
+ s3_conn = S3Hook(aws_conn_id=self.aws_conn_id, verify=self.verify)
+ data_df = mysql_hook.get_pandas_df(self.query)
+ self.log.info("Data from MySQL obtained")
+
+ self._fix_int_dtypes(data_df)
+ with tempfile.NamedTemporaryFile(mode='r+', suffix='.csv') as tmp_csv:
+ tmp_csv.file.write(data_df.to_csv(**self.pd_csv_kwargs))
+ tmp_csv.file.seek(0)
Review comment:
Thank you very much for your suggestions.
I already tried to use just to_csv and I failed. Also, at the beginning, I was using StringIO (equivalent to BytesIO that s3 hook uses) and I was suggested to use the way I am using right now. Everything can be found in this discussion:
https://github.com/apache/airflow/pull/6670#discussion_r361645289
----------------------------------------------------------------
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 #9054: Add MySqlToS3Operator
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #9054:
URL: https://github.com/apache/airflow/pull/9054#discussion_r437208286
##########
File path: airflow/providers/amazon/aws/operators/mysql_to_s3.py
##########
@@ -0,0 +1,126 @@
+#
+# 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 tempfile
+from typing import Optional, Union
+
+import numpy as np
+import pandas as pd
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.s3 import S3Hook
+from airflow.providers.mysql.hooks.mysql import MySqlHook
+from airflow.utils.decorators import apply_defaults
+
+
+class MySQLToS3Operator(BaseOperator):
+ """
+ Saves data from an specific MySQL query into a file in S3.
+
+ :param query: the sql query to be executed. If you want to execute a file, place the absolute path of it,
+ ending with .sql extension.
+ :type query: str
+ :param s3_bucket: bucket where the data will be stored
+ :type s3_bucket: str
+ :param s3_key: desired key for the file. It includes the name of the file
+ :type s3_key: str
+ :param mysql_conn_id: reference to a specific mysql database
+ :type mysql_conn_id: str
+ :param aws_conn_id: reference to a specific S3 connection
+ :type aws_conn_id: str
+ :param verify: Whether or not to verify SSL certificates for S3 connection.
+ By default SSL certificates are verified.
+ You can provide the following values:
+
+ - ``False``: do not validate SSL certificates. SSL will still be used
+ (unless use_ssl is False), but SSL certificates will not be verified.
+ - ``path/to/cert/bundle.pem``: A filename of the CA cert bundle to uses.
+ You can specify this argument if you want to use a different
+ CA cert bundle than the one used by botocore.
+ :type verify: bool or str
+ :param pd_csv_kwargs: arguments to include in pd.to_csv (header, index, columns...)
+ :type pd_csv_kwargs: dict
+ :param index: whether to have the index or not in the dataframe
+ :type index: str
+ :param header: whether to include header or not into the S3 file
+ :type header: bool
+ """
+
+ template_fields = ('s3_key', 'query',)
+ template_ext = ('.sql',)
+
+ @apply_defaults
+ def __init__(
+ self,
+ query: str,
+ s3_bucket: str,
+ s3_key: str,
+ mysql_conn_id: str = 'mysql_default',
+ aws_conn_id: str = 'aws_default',
+ verify: Optional[Union[bool, str]] = None,
+ pd_csv_kwargs: dict = None,
Review comment:
```suggestion
pd_csv_kwargs: Optional[dict] = None,
```
As we use `None` as default value
----------------------------------------------------------------
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] JavierLopezT commented on a change in pull request #9054: Add MySqlToS3Operator
Posted by GitBox <gi...@apache.org>.
JavierLopezT commented on a change in pull request #9054:
URL: https://github.com/apache/airflow/pull/9054#discussion_r437244791
##########
File path: airflow/providers/amazon/aws/operators/mysql_to_s3.py
##########
@@ -0,0 +1,126 @@
+#
+# 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 tempfile
+from typing import Optional, Union
+
+import numpy as np
+import pandas as pd
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.s3 import S3Hook
+from airflow.providers.mysql.hooks.mysql import MySqlHook
+from airflow.utils.decorators import apply_defaults
+
+
+class MySQLToS3Operator(BaseOperator):
+ """
+ Saves data from an specific MySQL query into a file in S3.
+
+ :param query: the sql query to be executed. If you want to execute a file, place the absolute path of it,
+ ending with .sql extension.
+ :type query: str
+ :param s3_bucket: bucket where the data will be stored
+ :type s3_bucket: str
+ :param s3_key: desired key for the file. It includes the name of the file
+ :type s3_key: str
+ :param mysql_conn_id: reference to a specific mysql database
+ :type mysql_conn_id: str
+ :param aws_conn_id: reference to a specific S3 connection
+ :type aws_conn_id: str
+ :param verify: Whether or not to verify SSL certificates for S3 connection.
+ By default SSL certificates are verified.
+ You can provide the following values:
+
+ - ``False``: do not validate SSL certificates. SSL will still be used
+ (unless use_ssl is False), but SSL certificates will not be verified.
+ - ``path/to/cert/bundle.pem``: A filename of the CA cert bundle to uses.
+ You can specify this argument if you want to use a different
+ CA cert bundle than the one used by botocore.
+ :type verify: bool or str
+ :param pd_csv_kwargs: arguments to include in pd.to_csv (header, index, columns...)
+ :type pd_csv_kwargs: dict
+ :param index: whether to have the index or not in the dataframe
+ :type index: str
+ :param header: whether to include header or not into the S3 file
+ :type header: bool
+ """
+
+ template_fields = ('s3_key', 'query',)
+ template_ext = ('.sql',)
+
+ @apply_defaults
+ def __init__(
+ self,
+ query: str,
+ s3_bucket: str,
+ s3_key: str,
+ mysql_conn_id: str = 'mysql_default',
+ aws_conn_id: str = 'aws_default',
+ verify: Optional[Union[bool, str]] = None,
+ pd_csv_kwargs: dict = None,
Review comment:
Thank you very much!. I made that change 11 days ago https://github.com/apache/airflow/pull/9054/commits/0336eaabf0c7e40e867995f81d5ea1952f9b992d I guess I was missing something else, cause now it works
----------------------------------------------------------------
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] JavierLopezT commented on pull request #9054: Add MySqlToS3Operator
Posted by GitBox <gi...@apache.org>.
JavierLopezT commented on pull request #9054:
URL: https://github.com/apache/airflow/pull/9054#issuecomment-640061410
Thank you very much @turbaszek
Regarding the static cheks, I have already tried some things for solving `Incompatible default for argument pd_csv_kwargs` but I am out of ideas. Could you (or anyone) guide me a little bit please?
Thank you very much in advance and sorry for the inconveniences
----------------------------------------------------------------
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 #9054: Add MySqlToS3Operator
Posted by GitBox <gi...@apache.org>.
feluelle commented on a change in pull request #9054:
URL: https://github.com/apache/airflow/pull/9054#discussion_r438028146
##########
File path: airflow/providers/amazon/aws/operators/mysql_to_s3.py
##########
@@ -0,0 +1,126 @@
+#
+# 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 tempfile
+from typing import Optional, Union
+
+import numpy as np
+import pandas as pd
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.s3 import S3Hook
+from airflow.providers.mysql.hooks.mysql import MySqlHook
+from airflow.utils.decorators import apply_defaults
+
+
+class MySQLToS3Operator(BaseOperator):
+ """
+ Saves data from an specific MySQL query into a file in S3.
+
+ :param query: the sql query to be executed. If you want to execute a file, place the absolute path of it,
+ ending with .sql extension.
+ :type query: str
+ :param s3_bucket: bucket where the data will be stored
+ :type s3_bucket: str
+ :param s3_key: desired key for the file. It includes the name of the file
+ :type s3_key: str
+ :param mysql_conn_id: reference to a specific mysql database
+ :type mysql_conn_id: str
+ :param aws_conn_id: reference to a specific S3 connection
+ :type aws_conn_id: str
+ :param verify: Whether or not to verify SSL certificates for S3 connection.
+ By default SSL certificates are verified.
+ You can provide the following values:
+
+ - ``False``: do not validate SSL certificates. SSL will still be used
+ (unless use_ssl is False), but SSL certificates will not be verified.
+ - ``path/to/cert/bundle.pem``: A filename of the CA cert bundle to uses.
+ You can specify this argument if you want to use a different
+ CA cert bundle than the one used by botocore.
+ :type verify: bool or str
+ :param pd_csv_kwargs: arguments to include in pd.to_csv (header, index, columns...)
+ :type pd_csv_kwargs: dict
+ :param index: whether to have the index or not in the dataframe
+ :type index: str
+ :param header: whether to include header or not into the S3 file
+ :type header: bool
+ """
+
+ template_fields = ('s3_key', 'query',)
+ template_ext = ('.sql',)
+
+ @apply_defaults
+ def __init__(
+ self,
+ query: str,
+ s3_bucket: str,
+ s3_key: str,
+ mysql_conn_id: str = 'mysql_default',
+ aws_conn_id: str = 'aws_default',
+ verify: Optional[Union[bool, str]] = None,
+ pd_csv_kwargs: Optional[dict] = None,
+ index: Optional[bool] = False,
+ header: Optional[bool] = False,
+ *args, **kwargs) -> None:
+ super().__init__(*args, **kwargs)
+ self.query = query
+ self.s3_bucket = s3_bucket
+ self.s3_key = s3_key
+ self.mysql_conn_id = mysql_conn_id
+ self.aws_conn_id = aws_conn_id
+ self.verify = verify
+
+ if not pd_csv_kwargs:
+ self.pd_csv_kwargs = {}
+ else:
+ self.pd_csv_kwargs = pd_csv_kwargs
+ if "index" not in self.pd_csv_kwargs:
+ self.pd_csv_kwargs["index"] = index
+ if "header" not in self.pd_csv_kwargs:
+ self.pd_csv_kwargs["header"] = header
+
+ def _fix_int_dtypes(self, df):
+ """
+ Mutate DataFrame to set dtypes for int columns containing NaN values."
+ """
+ for col in df:
+ if "float" in df[col].dtype.name and df[col].hasnans:
+ # inspect values to determine if dtype of non-null values is int or float
+ notna_series = df[col].dropna().values
+ if np.isclose(notna_series, notna_series.astype(int)).all():
+ # set to dtype that retains integers and supports NaNs
+ df[col] = np.where(df[col].isnull(), None, df[col]).astype(pd.Int64Dtype)
+
+ def execute(self, context):
+ mysql_hook = MySqlHook(mysql_conn_id=self.mysql_conn_id)
+ s3_conn = S3Hook(aws_conn_id=self.aws_conn_id, verify=self.verify)
+ data_df = mysql_hook.get_pandas_df(self.query)
+ self.log.info("Data from MySQL obtained")
+
+ self._fix_int_dtypes(data_df)
+ with tempfile.NamedTemporaryFile(mode='r+', suffix='.csv') as tmp_csv:
+ tmp_csv.file.write(data_df.to_csv(**self.pd_csv_kwargs))
+ tmp_csv.file.seek(0)
Review comment:
okay, but **Solution 1** should work. You should pass the file name not the object. Can you try this? `data_df.to_csv(tmp_csv.name, **self.pd_csv_kwargs)`
**EDIT:**
I looked up BytesIO again. BytesIO / StringIO does not create a file - only a buffer. So the "file" mentioned in your comment does not exist.
----------------------------------------------------------------
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