Way to automatically create SnowFlake table based on inferred field types from API Endpoint? (Python)

user_9682236 0
Viewed 825

Say I have a dataframe that has a row like:

{'ID':'123245','Comment':'This is my longer comment','Tax':1.07,'Units':2.0}

Is there a way in Python to do something like:

max([len(str(i)) for i in set(df['Comments'])])

And infer the max varchar and other metadata that I could then construct a SQL query to create that table (in my case, for SnowFlake)?

Since it would take additional logic not mentioned (e.g. try to cast as int, float, datetime, etc.), perhaps this is commonly done in an existing library.

Right now, it takes me some time for each endpoint to manually check across the fields and infer how to make each table in Snowflake, again, manually. Would like to automate this process.

Of course, one aspect of automating this without something more sophisticated like a library is that your max fields now (such as a comment that's 199 characters long) will likely be soon violated by future inputs into those fields if not, say, rounded up to a 'max' varchar such as telling such an algorithm a minimum varchar when it can't convert to float/int/date/etc.

snowflake-schema snowflake-cloud-data-platform endpoint sqlalchemy python
1 Answers

First off, as mentioned in the Snowflake docs, explicitly setting the maximum length of a VARCHAR column has no impact on performance and storage, so don't bother with that.

Regarding your general question, you can use their native Python connector to simply upload the DataFrame to your environment. Matching Python types to Snowflake types is done automatically.

If you want to only create the table without inserting data, upload df.iloc[:0]. And if you want to get the create table SQL, you can use get_ddl. Below is an example implementation.

import pandas as pd
import snowflake.connector
from snowflake.connector.pandas_tools import pd_writer
from snowflake.sqlalchemy import URL
import sqlalchemy

credentials = {**your_snowflake_credentials}

# Create example DataFrame
data = {
    "ID": "123245",
    "COMMENT": "This is my longer comment",
    "TAX": 1.07,
    "UNITS": 2,
}
df = pd.DataFrame([data])

# Upload empty DataFrame
df.iloc[:0].to_sql(
    "test_table",
    sqlalchemy.create_engine(URL(**credentials)),
    index=False,
    method=pd_writer,
)

# Retrieve the CREATE TABLE statement and drop the temporary table
# (if you really want to)
sql = "select get_ddl('table', 'test_table')"
with snowflake.connector.connect(**credentials) as connection:
    with connection.cursor() as cursor:
        create_table_sql = cursor.execute(sql).fetchone()[0]
        cursor.execute("drop table test_table")

print(create_table_sql)

Output:

CREATE OR REPLACE TABLE TEST_TABLE (
        ID VARCHAR(16777216),
        COMMENT VARCHAR(16777216),
        TAX FLOAT,
        UNITS NUMBER(38,0)
);
user_96810810
Related