如何在Java中无需文件IO从Parquet输入流提取数据？

无法在不创建临时文件的情况下，从包含Parquet文件数据的Java InputStream中提取数据，需要编写无需创建临时文件即可读取Parquet输入流数据的方法。目前已有可运行的代码，但必须依赖文件，请问如何修改实现无文件读取？

现有代码：

public static void main(String[] args) throws FileNotFoundException {
    SpringApplication.run(ParquetstreamApplication.class, args);
    Path path=new Path(Paths.get("").toAbsolutePath().toString()+"\\src\\main\\resources\\test-parquet.parquet");
    InputStream fileInputStream = new FileInputStream(path.toString());
    List<GenericRecord> recordList = new ArrayList<>();
     try (ParquetFileReader parquetFileReader = new ParquetFileReader(HadoopInputFile.fromPath(path, new Configuration()), ParquetReadOptions.builder().build())) {
            final ParquetMetadata footer = parquetFileReader.getFooter();
            final MessageType schema = createdParquetSchema(footer);
            PageReadStore pages;
            while ((pages = parquetFileReader.readNextRowGroup()) != null) {
                final long rows = pages.getRowCount();
                final MessageColumnIO columnIO = new ColumnIOFactory().getColumnIO(schema);
                final RecordReader<Group> recordReader = columnIO.getRecordReader(pages, new GroupRecordConverter(schema));

                for (int row = 0; row < rows; row++) {

                        final Map<String, Object> eventData = new HashMap<>();

                        int fieldIndex = 0;
                        final SimpleGroup simpleGroup = (SimpleGroup) recordReader.read();
                        GenericRecord record = new GenericData.Record(parseSchema());
                        for (Type field : schema.getFields()) {

                            try {
                               //System.out.print(field.getName()+": "+simpleGroup.getValueToString(fieldIndex, 0)+" ");
                                record.put(field.getName(),simpleGroup.getValueToString(fieldIndex, 0));
                            }
                            catch (Exception parquetException){
                               System.out.println("error");
                            }

                            fieldIndex++;

                        }
                        //System.out.println();
                    recordList.add(record);
                }
            }

要实现无文件读取Parquet输入流，核心是把InputStream转换成字节数组，然后使用Parquet提供的ByteArrayInputFile替代依赖文件路径的HadoopInputFile来构建ParquetFileReader。具体修改如下：

步骤说明 #

1. 将InputStream转换为字节数组：把输入流的内容读取到字节数组中，可借助工具类或手动实现。
2. 替换ParquetFileReader初始化方式：用ByteArrayInputFile包装字节数组，直接构建读取器，无需依赖物理文件。
3. 复用原有读取逻辑：Schema解析、行组读取、Record转换的核心逻辑可以保留。

修改后的完整代码示例 #

import org.apache.commons.io.IOUtils;
import org.apache.parquet.bytes.ByteArrayInputFile;
import org.apache.parquet.hadoop.ParquetFileReader;
import org.apache.parquet.io.InputFile;
import org.apache.parquet.schema.MessageType;
import org.apache.parquet.schema.Type;
import org.apache.parquet.io.api.RecordReader;
import org.apache.parquet.hadoop.metadata.ParquetMetadata;
import org.apache.parquet.column.page.PageReadStore;
import org.apache.parquet.column.ColumnIOFactory;
import org.apache.parquet.column.impl.MessageColumnIO;
import org.apache.parquet.example.data.Group;
import org.apache.parquet.example.data.simple.SimpleGroup;
import org.apache.parquet.example.data.convert.GroupRecordConverter;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

import java.io.FileInputStream;
import java.io.InputStream;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;

@SpringBootApplication
public class ParquetstreamApplication {

    public static void main(String[] args) throws Exception {
        SpringApplication.run(ParquetstreamApplication.class, args);
        Path path = Paths.get("").toAbsolutePath().resolve("src/main/resources/test-parquet.parquet");
        InputStream fileInputStream = new FileInputStream(path.toString());
        
        // 将InputStream转换为字节数组
        byte[] parquetBytes = IOUtils.toByteArray(fileInputStream);
        
        List<GenericRecord> recordList = new ArrayList<>();
        
        // 使用ByteArrayInputFile构建ParquetFileReader，无需临时文件
        InputFile inputFile = new ByteArrayInputFile(parquetBytes);
        try (ParquetFileReader parquetFileReader = ParquetFileReader.open(inputFile)) {
            final ParquetMetadata footer = parquetFileReader.getFooter();
            final MessageType schema = createdParquetSchema(footer);
            PageReadStore pages;
            
            while ((pages = parquetFileReader.readNextRowGroup()) != null) {
                final long rows = pages.getRowCount();
                final MessageColumnIO columnIO = new ColumnIOFactory().getColumnIO(schema);
                final RecordReader<Group> recordReader = columnIO.getRecordReader(pages, new GroupRecordConverter(schema));
                
                for (int row = 0; row < rows; row++) {
                    final SimpleGroup simpleGroup = (SimpleGroup) recordReader.read();
                    GenericRecord record = new GenericData.Record(parseSchema());
                    int fieldIndex = 0;
                    
                    for (Type field : schema.getFields()) {
                        try {
                            record.put(field.getName(), simpleGroup.getValueToString(fieldIndex, 0));
                        } catch (Exception parquetException) {
                            System.out.println("字段读取错误: " + field.getName());
                        }
                        fieldIndex++;
                    }
                    recordList.add(record);
                }
            }
        }
    }

    // 保留原有Schema解析方法
    private static MessageType createdParquetSchema(ParquetMetadata footer) {
        return footer.getFileMetaData().getSchema();
    }

    // 保留原有Avro Schema解析方法
    private static org.apache.avro.Schema parseSchema() {
        // 替换为你的实际Schema解析逻辑
        return null;
    }
}

补充说明 #

• 无额外依赖实现流转字节数组：如果不想引入Apache Commons IO，可以手动实现流转换逻辑：

private static byte[] inputStreamToByteArray(InputStream is) throws Exception {
    ByteArrayOutputStream buffer = new ByteArrayOutputStream();
    int nRead;
    byte[] data = new byte[1024];
    while ((nRead = is.read(data, 0, data.length)) != -1) {
        buffer.write(data, 0, nRead);
    }
    buffer.flush();
    return buffer.toByteArray();
}

• 版本兼容：ByteArrayInputFile适用于Parquet 1.12及以上版本，若使用旧版本，可尝试用SeekableByteArrayInputStream结合HadoopInputFile.fromStream方法实现，但需要处理可定位流的逻辑。

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