如何基于列筛选条件获取S3存储桶中的Parquet文件列表？

Great question! Let's break down how to get the list of Parquet files in your S3 bucket that match your deptno=2 and specific transactiontime criteria. Here are a few practical approaches you can use:

1. Use AWS Athena (No/Low-Code Solution) #

This is perfect if you want to avoid writing custom code and leverage AWS's managed query service to scan your S3 Parquet files.

Steps: #

• Create an Athena Table: First, define a table that maps to your S3 Parquet files and their schema. Run this DDL in the Athena query editor:

  CREATE EXTERNAL TABLE IF NOT EXISTS your_parquet_table (
      id STRING,
      name STRING,
      address STRING,
      zipcode STRING,
      deptno INT,
      transactiontime STRING
  )
  STORED AS PARQUET
  LOCATION 's3://bucket/folder/'
  TBLPROPERTIES ('parquet.compress'='SNAPPY');
  

• Run a Filter Query: Use Athena's built-in $path field to get the S3 paths of files containing matching rows. The query will automatically scan only relevant files (thanks to Parquet's columnar storage and stats):

  SELECT DISTINCT "$path" 
  FROM your_parquet_table 
  WHERE deptno = 2 
    AND transactiontime = '2019-10-24T21:14:39.503Z';
  

• Export Results: You can either copy the file paths directly from the Athena results pane or save the query output to another S3 bucket for later use.

2. Python Script (Flexible, Code-Based Solution) #

If you need to integrate this logic into an application or workflow, use Python with Parquet-processing libraries.

Option A: PyArrow + S3FS (High Efficiency) #

PyArrow is optimized for large-scale Parquet handling and can leverage file metadata to skip non-matching files quickly.

1. Install dependencies:

   pip install pyarrow s3fs
   

2. Script example:

   import pyarrow.parquet as pq
   import s3fs
   
   # Initialize connection to S3
   s3 = s3fs.S3FileSystem()
   s3_folder_path = "s3://bucket/folder/"
   
   # Create a dataset to handle all Parquet files in the folder
   parquet_dataset = pq.ParquetDataset(s3_folder_path, filesystem=s3)
   
   # Define your filter conditions
   filter_conditions = (
       parquet_dataset.schema.field("deptno") == 2
       & parquet_dataset.schema.field("transactiontime") == "2019-10-24T21:14:39.503Z"
   )
   
   # Get all matching files
   matching_files = []
   for fragment in parquet_dataset.get_fragments(filter=filter_conditions):
       matching_files.extend(fragment.files)
   
   # Print results
   print("Matching S3 files:")
   for file in matching_files:
       print(file)
   

Option B: Pandas + S3FS (Smaller Datasets) #

Great for smaller volumes of data where simplicity is key:

1. Install dependencies:

   pip install pandas pyarrow s3fs
   

2. Script example (read files one by one to save memory):

   import pandas as pd
   import s3fs
   
   s3 = s3fs.S3FileSystem()
   s3_parquet_paths = s3.glob("s3://bucket/folder/*.parquet")
   
   matching_files = []
   for file_path in s3_parquet_paths:
       # Read a single Parquet file
       df = pd.read_parquet(f"s3://{file_path}", filesystem=s3)
       # Check if the file contains any matching rows
       if not df[(df["deptno"] == 2) & (df["transactiontime"] == "2019-10-24T21:14:39.503Z")].empty:
           matching_files.append(file_path)
   
   print("Matching S3 files:")
   for file in matching_files:
       print(file)
   

Key Recommendations #

• Enable Parquet Statistics: When writing Parquet files (e.g., with Spark), ensure you enable metadata statistics. This allows tools like Athena and PyArrow to skip entire files that can't possibly match your filter, drastically speeding up queries.
• Check IAM Permissions: Make sure your AWS identity has s3:ListBucket and s3:GetObject permissions for the target bucket. If using Athena, you'll also need permissions for Athena execution and Glue Data Catalog (if using it).
• Optimize the Transaction Time Field: If you frequently query by transactiontime, consider storing it as a TIMESTAMP type instead of a string. This enables more flexible range queries (e.g., transactiontime BETWEEN '2019-10-24' AND '2019-10-25') and better query performance.
• Avoid Full File Scans: For large datasets, always prefer methods that use Parquet metadata (like Athena or PyArrow Dataset) over reading entire files into memory.

内容的提问来源于stack exchange，提问作者Govinda