Aws-data-wrangler: Copy parquet files then query them with Athena

[JPFrancoia]

Hi,
I use aws-data-wrangler to process pandas dataframes. Once they are processed, I export them to parquet files with:

    wr.pandas.to_parquet(
        dataframe=my_dataframe,
        description=DESCRIPTION,
        columns_comments=COLUMN_COMMENTS,
        parameters=DATASET_TAGS,
        database=my_database,
        table=f"{table}_{latest_refresh_date}",
        path=f"s3://{bucket_out}/{sub_path}/{latest_refresh_date}/",
        procs_cpu_bound=1,
        partition_cols=["date"],
        mode="overwrite_partitions",
        preserve_index=False,
    )

By doing it this way, it also creates a Glue table. The parquet files happily live in a S3 bucket, and I can query the data with Athena using the name of the Glue table, like this:

select * from {table}_{latest_refresh_date}

Now let's say that I get new data. The new data should be stored in a new S3 path: s3://{bucket_out}/{sub_path}/{other_refresh_date}/ for example. I process the new data as before, but I don't want to re-process the old data. So I copy the parquet files from the old path to the new one, with a simple S3 copy. The old data isn't handled with wr.pandas.to_parquet.

Now when I want to query the data living in s3://{bucket_out}/{sub_path}/{other_refresh_date}/, I can only access the new data.

select * from {table}_{other_refresh_date}

It seems I can only query the data that was added to the Glue table. I naively thought that Athena would query a S3 path but apparently it's more complicated than that.

Could you explain to me why this is happening, and maybe suggest a work around?
Do I need to somehow register the old files to the new Athena table?

[igorborgest]

Hi @JPFrancoia!

Amazon Athena is able to query new files on s3 w/o any additional work, but the same is not valid for new partitions.

When new partitions directories are added on s3 you need to also add the related metadata in the Glue Catalog.

Probably the easiest way to do it is running this query on Athena:

MSCK REPAIR TABLE {table}_{other_refresh_date};

You also can do it programatically through Wrangler:

wr.athena.repair_table(table="...", database="...")

Please, let me know if it will be enough for you.

[JPFrancoia]

Hi @igorborgest , thanks for your quick answer.

I was able to add partition directories with repair_table, thanks. Naive question though: the "repair table" approach feels like a workaround to me. Is it really its intended use? Is it a durable approach?

I managed to get the same result by doing wr.pandas.read_table (reading from the previous table) and then wr.pandas.to_parquet (writing to the new table). It feels a bit convoluted though, since it does parquet > dataframe > parquet.

Dropping the partitioning also gives me what I want out of the box.

What do you think is the best approach?

[igorborgest]

@JPFrancoia

The repair_table is good enough to keep in productive ETLs and has the benefits to be simple and to guarantee that no partitions will be forgotten.

But you are right, there are other better options that demand a few more development work, though.

Let me explain two other different approaches:

1 - Add the specific partition via Athena query:

ALTER TABLE orders ADD PARTITION (dt = '2016-05-14', country = 'IN');

Ref: https://docs.aws.amazon.com/athena/latest/ug/alter-table-add-partition.html

Through Wrangler:

wr.athena.query(
    database="...",
    query="ALTER TABLE orders ADD PARTITION..."
)

2- Add the specific partition via Glue Catalog:

The fastest and more complex approach. You can add the partition via the Glue Catalog API.

Ref0: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/glue.html#Glue.Client.create_partition
Ref1: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/glue.html#Glue.Client.batch_create_partition

Through Wrangler:

Today the function that should help on that is so tricky that I will not recommend it by now. But we will add a far away better version on version 1.0.0.

The link for the current implementation (NOT a recommendation): https://github.com/awslabs/aws-data-wrangler/blob/0.3.2/awswrangler/glue.py#L228

[cfregly]

Great info! Is there a partition limit that Athena users should be aware of? I believe I hit a limit recently when trying to do > 100 partitions, but I can’t remember the exact details at the moment.

[igorborgest]

Exactly @cfregly, the BatchCreatePartition method in the AWS Glue API only supports a maximum of 100 partitions per request (Ref). But is possible to reach up to 10MM partitions per table with multiples requests (Ref).

@JPFrancoia I've just pushed the functions that will handle the partitions for parquet tables to the branch dev-1.0.0. Please, take a look and give us some feedback if you have some.

wr.s3.to_parquet():
source
docs
obs: Now the output of this functions will return the partitions_values that could be used to add the partitions to any other table independently with the second function bellow

wr.catalog.add_parquet_partitions():
source
docs

To get your hand dirty and test this development version:

pip install git+https://github.com/awslabs/[email protected]

P.S. Docs feedbacks will be also very welcome.

[JPFrancoia]

Oh, I see the to_parquet function is moving from the pandas module to the s3 module. Makes sense.

I do have an observation/question. When I batch-process data for {latest_refresh_date}, I would use the wr.s3.to_parquet() function to export my dataframe to parquet, let's say with a partition on date.

I will get back the partitions_values created during the export, that's nice, but at this point in time I wouldn't do anything with them (at least in my current workflow).

When I receive a new batch of data on {other_refresh_date} (could be days/months since the last refresh), I will want to copy the S3 objects from the last refresh into the new refresh S3 path. Once the copy is done, I will want to register the partitions to the catalog with wr.catalog.add_parquet_partitions(). But at this point, I would want to query the partitions from the old refresh {latest_refresh_date} and then add them to the {other_refresh_date} table.

Basically this is what I envision:

# ...Copy S3 files from s3://{bucket_out}/{sub_path}/{latest_refresh_date}/
#   to s3://{bucket_out}/{sub_path}/{other_refresh_date}/...

old_partitions = wr.catalog.get_parquet_partitions(table="{table}_{latest_refresh_date}"):

# ...Create {table}_{other_refresh_date} in the Glue catalog

wr.catalog.add_parquet_partitions(table=f"{table}_{other_refresh_date}", partitions=old_partitions)

# ...Process the new data...

Does it make sense?

[igorborgest]

Hi @JPFrancoia,

Take a look in the fresh new wr.catalog.get_parquet_partitions() function (source)(docs).
Is this the last missing piece right?

Btw, thinking more broadly about your challenge, why not just add the new files to the same table using the mode="append" or mode="partition_upsert"?

[JPFrancoia]

Take a look in the fresh new wr.catalog.get_parquet_partitions() function (source)(docs).
Is this the last missing piece right?

That's perfect, that's exactly what I need!

Btw, thinking more broadly about your challenge, why not just add the new files to the same table using the mode="append" or mode="partition_upsert"?

Because each refresh date ({latest_refresh_date}, {other_refresh_date}, etc) corresponds to the date we receive a new batch of data (== a version). And for each new batch, I can potentially receive files that update "past" data:

• on {latest_refresh_date}, I can have a file file1.csv with values A, B, and C inside
• on {other_refresh_date}, I can receive a new file file1.csv with values B, B, and C inside (value A gets updated).

If I don't receive a new file file1.csv on {other_refresh_date}, it means the corresponding values don't need to be updated.

So basically, with carefully chosen partitions and the overwrite_partition mode, I know I can copy past partitions and process the new file1.csv: if the data in the file overwrites past partitions, I know I'll ultimately get the most up to data data in the table {my_dataset}_{other_refresh_date}.

And I need to keep the data from {my_dataset}_{latest_refresh_date} because our data science team might need to pull this version of the data again.

Out of curiosity, what does partition_upsert do? It's not in the doc of the dev API.

[igorborgest]

@JPFrancoia it makes sense, now I see.

partition_upsert is the new alias for overwrite_partition in the version 1.0.0. We thought it would be more intuitive.
But I not sure about it... What do you think?

[JPFrancoia]

partition_upsert is the new alias for overwrite_partition in the version 1.0.0. We thought it would be more intuitive.
But I not sure about it... What do you think?

I don't have a super strong opinion about it. I think if you know what an upsert is (which wasn't my case 5 min ago haha) it makes sense. That said, I'm not sure it's more intuitive, and it also doesn't carry as clearly the meaning of "be careful, data might be overwritten and you might loose it". But that's just how I see it.