suggest to add pandas.to_sql()

[josecw]

Is it possible to have ability similar to function below:

gluecontext.write_dynamic_frame.from_jdbc_conf() as below?
datasink4 = glueContext.write_dynamic_frame.from_jdbc_conf(frame = datasource0, catalog_connection = "test_red", connection_options = {"preactions":"truncate table target_table;","dbtable": "target_table", "database": "redshiftdb"}, redshift_tmp_dir = 's3://s3path', transformation_ctx = "datasink4")

Currently the way we do is:

1. Get SQL from S3 file and pass into pandas.read_sql_athena()
2. use SQLAlchemy to execute preactions SQL. For our case is delete before load
3. use SQLAlchemy and pandas.to_sql() to append dataframe into aurora table

[igorborgest]

Hi @BellyTheMagnificent! Thanks again for one more valuable input!

We just developed a Aurora+Pandas integration available on version +0.1.4.

Now you can LOAD (MySQL/PostgreSQL) and UNLOAD (MySQL) to/from Pandas DataFrames using the HIGH throughput S3 integration:

import awswrangler as wr

wr.pandas.to_aurora(
    dataframe=df,
    connection=con,
    schema="...",
    table="..."
)

df = wr.pandas.read_sql_aurora(
    sql="SELECT ...",
    connection=con
)

For those operation and databases that does not have S3 integration, please keep using the native Pandas/PySpark APIs:

1. https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.DataFrame.to_sql.html
2. https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.read_sql.html
3. https://spark.apache.org/docs/latest/api/python/pyspark.sql.html?highlight=jdbc#pyspark.sql.DataFrameReader.jdbc
4. https://spark.apache.org/docs/latest/api/python/pyspark.sql.html?highlight=jdbc#pyspark.sql.DataFrameWriter.jdbc

Please, test it and give us some feedbacks! 🚀

[josecw]

Encoutered issue below:
pg8000.core.ProgrammingError: {'S': 'ERROR', 'V': 'ERROR', 'C': '22P04', 'M': 'unterminated CSV quoted field', 'W': 'COPY table_A, line 1804405: ""93215","XXXXX","I","AA","93215","X","XXXXX","XXX","\\","2020-01-05 00:00:00.000","XXXX","..."\nSQL statement "copy [schema].[table] from \'/rdsdbdata/extensions/aws_s3/amazon-s3-fifo-29761-20200106T092927Z-0\' with (FORMAT CSV, DELIMITER \',\', QUOTE \'"\', ESCAPE \'\\\')"', 'F': 'copy.c', 'L': '4530', 'R': 'CopyReadAttributesCSV'}

Wonder if this is cause by the value \ in one of the column.

Still testing. It would be great if it can:

1. Perform rollback when process fail
   pg8000.core.ProgrammingError: {'S': 'ERROR', 'V': 'ERROR', 'C': '25P02', 'M': 'current transaction is aborted, commands ignored until end of transaction block', 'F': 'postgres.c', 'L': '1356', 'R': 'exec_parse_message'}
2. Option to pass Glue Connection Name as connection

[igorborgest]

Hi @BellyTheMagnificent!
Really valuable feedbacks, thank you!

I just implemented all 3 points raised by you in the above PR. What you think?

1. Also accepts Glue connection name option instead of only PEP 249 connections
2. Rolling back in case of failure
3. Handling special characters (escaping)

Maybe the better way to you see how it work is through the added tests.

[josecw]

It look great. Thank you for the help!

Tried replace() before to_aurora() but it don't work too

str_cols = df.select_dtypes(include=['object']).columns df.loc[:, str_cols] = df.loc[:, str_cols].replace('\\','\\\\')

[igorborgest]

@BellyTheMagnificent let me know if you have more feedbacks in the future!