Snowflake is not deducting partitioned by column in Parquet

user_13125042 0
Viewed 574

I have a question about Snowflake's new capability -Infer Schema table function. The INFER SCHEMA function performs admirably on the parquet file and returns the correct data type. However, when the parquet files are partitioned and stored in S3, the INFER SCHEMA does not function as it does with pyspark dataframes.

In DataFrames, the partition folder name and value are read as the last column; is there a way to achieve the same result in the Snowflake Infer schema?

Example:

enter image description here

@GregPavlik - The input is in structured parquet format. When the parquet files are stored in S3 without a partition, the schema is perfectly derived.

Example : { "AGMT_GID": 1714844883, "AGMT_TRANS_GID": 640481290, "DT_RECEIVED": "20 302", "LATEST_TRANSACTION_CODE": "I" }

The Snowflake infer schema provides me with 4 column names as well as their data types.

However if the parquet file is stored in partition - like shown in image above.

under - LATEST_TRANSACTION_CODE =I/ folder i would have the file as

Example : { "AGMT_GID": 1714844883, "AGMT_TRANS_GID": 640481290, "DT_RECEIVED": "20 302" }

In this case, snowflake infer Schema only provides three columns; however, reading the same file in Pyspark dataframe prints all four columns.

I'm wondering if there is a workaround in Snowflake to read a partitioned parquet file.

parquet snowflake-cloud-data-platform
1 Answers

Faced this issue with snowflake while handling partitioned parquet files. This problem happens not only in infer_schema., Following flows doesn't deduct partitioned by column as a column in snowflake:

  • COPY INTO TABLE from parquet
  • MERGE INTO TABLE from parquet
  • SELECT from parquet
  • INFER_SCHEMA from parquet

Snowflake treats parquet files as file and ignore meta information at folder names. Apache Spark deducts partitioned columns intelligently.

Following approaches are ways to handle it, till Snowflake Team handles this.

Approach 1

Handle this using Snowflake metadata features.

As of now, Snowflake metadata provide only

  • METADATA$FILENAME - Name of the staged data file the current row belongs to. Includes the path to the data file in the stage.
  • METADATA$FILE_ROW_NUMBER - Row number for each record

We could do something like this:

    select $1:normal_column_1, ..., METADATA$FILENAME  
        FROM
            '@stage_name/path/to/data/' (pattern => '.*.parquet')
    limit 5;

This will give a column with full path to the partitioned file. But we need to handle deducing the column from it. For example: it would give something like:

METADATA$FILENAME 
----------
path/to/data/year=2021/part-00020-6379b638-3f7e-461e-a77b-cfbcad6fc858.c000.snappy.parquet

We could do a regexp_replace and get the partition value as column like this:

    select 
        regexp_replace(METADATA$FILENAME, '.*\/year=(.*)\/.*', '\\1'
        ) as year
        $1:normal_column_1,  
    FROM
            '@stage_name/path/to/data/' (pattern => '.*.parquet')
    limit 5;
  • In the above regexp, we give the partition key.
  • Third parameter \\1 is the regex group match number. In our case, first group match - this holds the partition value.

Approach 2

If we have control over the flow that writes the source parquet file.

  • Add a duplicate column that has same content as partition by column content
  • This should happen before writing to parquet. So the parquet file will have this column content.
df.withColumn("partition_column", col("col1")).write.partitionBy("partition_column").parquet(path)
  • With this approach, if we do this once, in all the usage of the parquet (COPY, MERGE, SELECT, INFER), new column will start appearing.

Approach 3

If we do not have control over the flow that writes the source parquet file.

  • This approach is more domain specific and data model specific.
  • In many usecases, we need to reverse engineer how the partitioned by column is related to the data.
  • Can it be generated from other columns ? Let's say, if the data is partitioned by year, where year is the derived data from created_by column, then this derived data can be regenerated again.
  • Can it be generated by joining with another snowflake table ? Let's say parquet has an id which can be joined with another table to derive at our column dynamically

Approach 3 is more problem/domain specific. Also we need to handle this in all use cases of the parquet ( COPY, MERGE, SELECT etc).

user_12946670
Related