SQL to Amazon S3

Use SqlToS3Operator to copy data from a SQL server to an Amazon Simple Storage Service (S3) file. SqlToS3Operator is compatible with any SQL connection as long as the SQL hook has function that converts the SQL result to pandas dataframe (e.g. MySQL, Hive, …).

With the default df_type="pandas" the operator needs the pandas extra. Writing Parquet files (file_format="parquet") or reading with read_kwargs={"dtype_backend": "pyarrow"} also needs pyarrow, which the parquet extra installs: apache-airflow-providers-amazon[pandas,parquet].

Prerequisite Tasks

To use these operators, you must do a few things:

Operators

MySQL to Amazon S3 transfer operator

This example sends the response of a MySQL query to an Amazon S3 file.

To get more information about this operator visit: SqlToS3Operator

Example usage:

tests/system/amazon/aws/example_sql_to_s3.py[source]

sql_to_s3_task = SqlToS3Operator(
    task_id="sql_to_s3_task",
    sql_conn_id=conn_id_name,
    query=SQL_QUERY,
    s3_bucket=bucket_name,
    s3_key=key,
    replace=True,
)

Grouping

We can group the data in the table by passing the groupby_kwargs param. This param accepts a dict which will be passed to pandas groupby() as kwargs.

Example usage:

tests/system/amazon/aws/example_sql_to_s3.py[source]

sql_to_s3_task_with_groupby = SqlToS3Operator(
    task_id="sql_to_s3_with_groupby_task",
    sql_conn_id=conn_id_name,
    query=SQL_QUERY,
    s3_bucket=bucket_name,
    s3_key=key,
    replace=True,
    groupby_kwargs={"by": "color"},
)

Paging

A result that does not fit in memory is read in pages with the page_size param, the way GenericTransfer pages a source: one query per page through paginated_sql_statement_clause ("{} LIMIT {} OFFSET {}" by default, formatted with the query, the page size and the offset), one file per page (<key>_0, <key>_1, …), and only one page in memory at a time. The transfer stops at the first page that comes back short. On most databases the query needs an ORDER BY for the pages to be stable; a hook that serves pages itself, such as one streaming a single result, needs none. page_size cannot be combined with max_rows_per_file or groupby_kwargs.

Example usage:

tests/system/amazon/aws/example_sql_to_s3.py[source]

sql_to_s3_task_with_page_size = SqlToS3Operator(
    task_id="sql_to_s3_with_page_size_task",
    sql_conn_id=conn_id_name,
    query=SQL_QUERY,
    s3_bucket=bucket_name,
    s3_key=key,
    replace=True,
    page_size=2,
)

Reference

Was this entry helpful?