使用S3FileSystem glob方法读取Parquet文件时遇错误求助

问题场景 #

尝试使用S3FileSystem的glob方法读取S3指定日期文件夹下的所有Parquet文件，编写代码如下：

def read_parquet_files_from_s3(self, table, schema, start_date, tenant_id):
    bucket_name = 'mybucket'
    start_date_object = datetime.strptime(start_date, "%m/%d/%Y").date()
    date_str = start_date_object.strftime("%Y/%m/%d")
    folder_path = f"archive_test/{tenant_id}/{table}/{date_str}"
    aws_keys = awsLogin(self.app_id, self.s3_conn_id)
    session = boto3.session.Session(aws_access_key_id=aws_keys['access_key_id'],
                                    aws_secret_access_key=aws_keys['secret_access_key'],
                                    aws_session_token=aws_keys['session_token'])
    self.token_expiration = aws_keys['expiration']
    try:
        import s3fs
        s3 = s3fs.S3FileSystem(key=aws_keys['access_key_id'],
                               secret=aws_keys['secret_access_key'],
                               token=aws_keys['expiration'])
        s3_path = f"s3://{bucket_name}/{folder_path}"
        print(f"s3 path is : {s3_path}")
        parquet_files = s3.glob(f"{s3_path}/*.parquet")
        print(parquet_files)
        if not parquet_files:
            print(f"No Parquet files found in {s3_path}")
            return []
        print(f"Found {len(parquet_files)} Parquet files in {s3_path}")

运行时出现错误：

Error reading Parquet file : 'datetime.datetime' object has no attribute 'split'.

错误原因 #

代码中初始化S3FileSystem时，token参数传入了aws_keys['expiration']——这是一个datetime对象，但s3fs的token参数要求传入的是AWS会话令牌的字符串值，也就是aws_keys['session_token']。s3fs内部会对token参数执行字符串分割操作，传入datetime对象就会触发'datetime.datetime' object has no attribute 'split'的错误。

修正后的代码 #

将S3FileSystem初始化的token参数替换为aws_keys['session_token']即可：

def read_parquet_files_from_s3(self, table, schema, start_date, tenant_id):
    bucket_name = 'mybucket'
    start_date_object = datetime.strptime(start_date, "%m/%d/%Y").date()
    date_str = start_date_object.strftime("%Y/%m/%d")
    folder_path = f"archive_test/{tenant_id}/{table}/{date_str}"
    aws_keys = awsLogin(self.app_id, self.s3_conn_id)
    session = boto3.session.Session(aws_access_key_id=aws_keys['access_key_id'],
                                    aws_secret_access_key=aws_keys['secret_access_key'],
                                    aws_session_token=aws_keys['session_token'])
    self.token_expiration = aws_keys['expiration']
    try:
        import s3fs
        # 修正token参数，传入session_token字符串而非expiration datetime对象
        s3 = s3fs.S3FileSystem(key=aws_keys['access_key_id'],
                               secret=aws_keys['secret_access_key'],
                               token=aws_keys['session_token'])
        s3_path = f"s3://{bucket_name}/{folder_path}"
        print(f"s3 path is : {s3_path}")
        parquet_files = s3.glob(f"{s3_path}/*.parquet")
        print(parquet_files)
        if not parquet_files:
            print(f"No Parquet files found in {s3_path}")
            return []
        print(f"Found {len(parquet_files)} Parquet files in {s3_path}")

额外验证点 #

• 确认awsLogin返回的aws_keys字典中，session_token是有效的字符串类型令牌，expiration是datetime类型的过期时间，二者不要混淆。
• 若后续需要处理令牌过期逻辑，可使用self.token_expiration存储的datetime值来判断是否需要重新获取令牌。

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