SNOW-2372686: Write large_pandas_backend_df.to_snowflake() via parquet.

[sfc-gh-mvashishtha]

To implement to_snowflake() for a Snowpark pandas dataframe on the pandas backend with large enough data, upload data via a parquet file instead of via a Snowpark dataframe. The Snowpark dataframe typically inserts values through parametrized SQL queries.

Benchmarking showed that a good threshold to switch to parquet was roughly 3 MB, so I've set that as the configurable default switching threshold. Performance of this approach seems to improve with dataset size. Exporting an 800 MB dataframe took about 55 seconds via parquet versus about 429 seconds via the old method, so we get over 7x speedup.

This work took me much longer than expected, so I am sending out this PR with just the changes for DataFrame.to_snowflake(). We can address Series.to_snowflake() separately.

We can take a similar approach to speed up pandas_backend_df.move_to('snowflake').

benchmark script with a 3XL warehouse

import pandas as native_pd
import modin.pandas as pd
import snowflake.snowpark.modin.plugin
import numpy as np
import datetime
from snowflake.snowpark.session import Session
from tests.utils import Utils
from multiprocessing import Pool
import psutil



from time import perf_counter

def get_benchmark_result(input_data: native_pd.DataFrame, method: str):
    if method == "parquet":
        pd.session.pandas_to_snowflake_row_limit = 0
    else:
        pd.session.pandas_to_snowflake_row_limit = int(1e99)

    df = pd.DataFrame(input_data).set_backend('pandas').pin_backend()
    table = Utils.random_table_name()
    start = perf_counter()
    df.to_snowflake(table)
    end = perf_counter()
    Utils.drop_table(pd.session, table)
    return {
        "time": datetime.timedelta(seconds=end-start),
        "memory_usage": input_data.memory_usage().sum(),
        "num_rows": input_data.shape[0],
        "num_columns": input_data.shape[1],
        "method": method
    }



# num_processes = psutil.cpu_count(logical=False)
# for 
session = Session.builder.create()
rows = []
for input_data in (
    native_pd.DataFrame(np.random.randint(low=0,high=1000,size=(int(1e0), 1))),     
    native_pd.DataFrame(np.random.randint(low=0,high=1000,size=(int(1e3), 1))),    
    native_pd.DataFrame(np.random.randint(low=0,high=1000,size=(int(1e3), 3))),     
    native_pd.DataFrame(np.random.randint(low=0,high=1000,size=(int(1e4), 1))),
    native_pd.DataFrame(np.random.randint(low=0,high=1000,size=(int(1e4), 3))),    
    native_pd.DataFrame(np.random.randint(low=0,high=1000,size=(int(1e5), 1))),    
    native_pd.DataFrame(np.random.randint(low=0,high=1000,size=(int(1e5), 3))),     
    native_pd.DataFrame(np.random.randint(low=0,high=1000,size=(int(1e6), 1))),
    native_pd.DataFrame(np.random.randint(low=0,high=1000,size=(int(1e6), 3))),    
    native_pd.DataFrame(np.random.randint(low=0,high=1000,size=(int(1e6), 10))),
    native_pd.DataFrame(np.random.randint(low=0,high=1000,size=(int(1e7), 1))),
    native_pd.DataFrame(np.random.randint(low=0,high=1000,size=(int(1e7), 10))),    
):
    for method in ("parquet", "snowpark"):
        rows.append(get_benchmark_result(input_data, method=method))

native_pd.DataFrame(rows).to_parquet('to_snowflake_timing_2', index=False)

[sfc-gh-joshi]

Very cool! That's an impressive perf graph.

Left a few comments.