读取Azure存储BlockBlob中Parquet文件时遇‘非Parquet文件’错误的解决

错误原因分析 #

这个错误的核心原因有两个：

1. 输入流不支持随机访问：Parquet 文件需要读取头部和尾部的魔数（PAR1）来验证合法性，而 blockBlobClient.openInputStream() 返回的普通 InputStream 不支持 seek（随机跳转）操作，导致无法正确读取尾部魔数，返回了无效的 [0, 0, 0, 0]。
2. 潜在的文件/配置问题：可能是目标 Blob 为空、文件本身不是有效 Parquet 文件，或者客户端构建时的冗余配置引发了隐性问题。

解决方案 #

步骤1：先验证文件有效性 #

在修改代码前，先确认基础问题：

• 登录 Azure 存储账户，检查 data/first/test.parquet 是否存在、文件大小是否大于0。
• 将文件下载到本地，用 parquet-tools meta test.parquet 命令验证是否为有效 Parquet 文件（如果没有工具，也可以用 Spark/Pandas 尝试读取）。
• 确认路径、容器名称的拼写（Azure Blob 路径大小写敏感）。

步骤2：修复代码（两种可选方案） #

方案一：包装可Seek的输入流（无需依赖Hadoop）

手动实现支持 seek 的输入流，适配 ParquetReader 的需求：

import org.apache.parquet.io.InputFile;
import org.apache.parquet.io.SeekableInputStream;
import org.apache.parquet.avro.AvroParquetReader;
import org.apache.parquet.io.ParquetReader;
import org.apache.avro.generic.GenericRecord;
import com.azure.storage.blob.*;
import java.io.*;

public void parquetReader() throws IOException {
    // 移除冗余配置，只保留一种认证方式
    BlobServiceClient blobServiceClient = new BlobServiceClientBuilder()
        .connectionString(storageAccountConnectionString)
        .buildClient();

    BlobContainerClient blobContainerClient = blobServiceClient.getBlobContainerClient(containerName);
    String path = "data/first/test.parquet";
    BlockBlobClient blockBlobClient = blobContainerClient.getBlobClient(path).getBlockBlobClient();

    // 前置校验
    if (!blockBlobClient.exists()) {
        throw new FileNotFoundException("Parquet file not found: " + path);
    }
    long blobSize = blockBlobClient.getProperties().getBlobSize();
    if (blobSize == 0) {
        throw new IOException("Parquet file is empty: " + path);
    }

    try (InputStream inputStream = blockBlobClient.openInputStream()) {
        // 初始化时标记整个流，支持reset操作
        inputStream.mark((int) blobSize);

        // 实现支持Seek的输入流
        SeekableInputStream seekableStream = new SeekableInputStream() {
            @Override
            public long getPos() throws IOException {
                return blobSize - inputStream.available();
            }

            @Override
            public void seek(long pos) throws IOException {
                if (pos < 0 || pos > blobSize) {
                    throw new IOException("Invalid seek position: " + pos);
                }
                inputStream.reset();
                long skipped = 0;
                while (skipped < pos) {
                    long skip = inputStream.skip(pos - skipped);
                    if (skip == 0) {
                        throw new IOException("Failed to skip to position: " + pos);
                    }
                    skipped += skip;
                }
            }

            @Override
            public int read() throws IOException {
                return inputStream.read();
            }

            @Override
            public int read(byte[] b, int off, int len) throws IOException {
                return inputStream.read(b, off, len);
            }
        };

        // 构建InputFile实例
        InputFile inputFile = new InputFile() {
            @Override
            public long getLength() throws IOException {
                return blobSize;
            }

            @Override
            public SeekableInputStream newStream() throws IOException {
                return seekableStream;
            }
        };

        // 读取Parquet文件
        try (ParquetReader<GenericRecord> reader = AvroParquetReader.<GenericRecord>builder(inputFile).build()) {
            GenericRecord record;
            while ((record = reader.read()) != null) {
                // 处理读取到的记录
                System.out.println(record);
            }
        }
    }
}

方案二：使用Hadoop Azure集成（更简洁推荐）

如果项目已经依赖Hadoop，直接用Hadoop的FileSystem访问Azure Blob，Parquet原生支持该方式：

1. 添加Maven依赖：

<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-azure</artifactId>
    <version>3.3.6</version>
</dependency>

2. 修改代码：

import org.apache.parquet.avro.AvroParquetReader;
import org.apache.parquet.io.ParquetReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import java.io.IOException;

public void parquetReader() throws IOException {
    Configuration conf = new Configuration();
    // 配置Azure存储密钥
    conf.set("fs.azure.account.key." + storageAccountName + ".blob.core.windows.net", blobKey);
    
    // 构建Blob路径（格式：wasbs://容器名@存储账户名.blob.core.windows.net/文件路径）
    Path parquetPath = new Path("wasbs://" + containerName + "@" + storageAccountName + ".blob.core.windows.net/data/first/test.parquet");
    
    // 直接读取Parquet文件
    try (ParquetReader<GenericRecord> reader = AvroParquetReader.<GenericRecord>builder(parquetPath).build()) {
        GenericRecord record;
        while ((record = reader.read()) != null) {
            // 处理读取到的记录
            System.out.println(record);
        }
    }
}

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