问：在内存中将 Avro 转换为 Parquet答案

【问题标题】：Q: Converting Avro to Parquet in Memory问：在内存中将 Avro 转换为 Parquet
【发布时间】：2016-09-22 06:31:47
【问题描述】：

我正在接收来自 Kafka 的 Avro 记录。我想将这些记录转换为 Parquet 文件。我正在关注这篇博文：http://blog.cloudera.com/blog/2014/05/how-to-convert-existing-data-into-parquet/

到目前为止的代码大致是这样的：

final String fileName
SinkRecord record, 
final AvroData avroData

final Schema avroSchema = avroData.fromConnectSchema(record.valueSchema());
CompressionCodecName compressionCodecName = CompressionCodecName.SNAPPY;

int blockSize = 256 * 1024 * 1024;
int pageSize = 64 * 1024;

Path path = new Path(fileName);
writer = new AvroParquetWriter<>(path, avroSchema, compressionCodecName, blockSize, pageSize);

现在，这将执行 Avro 到 Parquet 的转换，但它会将 Parquet 文件写入磁盘。我想知道是否有一种更简单的方法可以将文件保存在内存中，这样我就不必管理磁盘上的临时文件。谢谢

【问题讨论】：

标签： java hadoop avro parquet

【解决方案1】：

请查看我的博客，https://yanbin.blog/convert-apache-avro-to-parquet-format-in-java/必要时翻译成英文

package yanbin.blog;
 
import org.apache.parquet.io.DelegatingPositionOutputStream;
import org.apache.parquet.io.OutputFile;
import org.apache.parquet.io.PositionOutputStream;
 
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.OutputStream;
 
public class InMemoryOutputFile implements OutputFile {
    private final ByteArrayOutputStream baos = new ByteArrayOutputStream();
 
    @Override
    public PositionOutputStream create(long blockSizeHint) throws IOException { // Mode.CREATE calls this method
        return new InMemoryPositionOutputStream(baos);
    }
 
    @Override
    public PositionOutputStream createOrOverwrite(long blockSizeHint) throws IOException {
        return null;
    }
 
    @Override
    public boolean supportsBlockSize() {
        return false;
    }
 
    @Override
    public long defaultBlockSize() {
        return 0;
    }
 
    public byte[] toArray() {
        return baos.toByteArray();
    }
 
    private static class InMemoryPositionOutputStream extends DelegatingPositionOutputStream {
 
        public InMemoryPositionOutputStream(OutputStream outputStream) {
            super(outputStream);
        }
 
        @Override
        public long getPos() throws IOException {
            return ((ByteArrayOutputStream) this.getStream()).size();
        }
    }
}

    public static <T extends SpecificRecordBase> void writeToParquet(List<T> avroObjects) throws IOException {
        Schema avroSchema = avroObjects.get(0).getSchema();
        GenericData genericData = GenericData.get();
        genericData.addLogicalTypeConversion(new TimeConversions.DateConversion());
        InMemoryOutputFile outputFile = new InMemoryOutputFile();
        try (ParquetWriter<Object> writer = AvroParquetWriter.builder(outputFile)
                .withDataModel(genericData)
                .withSchema(avroSchema)
                .withCompressionCodec(CompressionCodecName.SNAPPY)
                .withWriteMode(ParquetFileWriter.Mode.CREATE)
                .build()) {
            avroObjects.forEach(r -> {
                try {
                    writer.write(r);
                } catch (IOException ex) {
                    throw new UncheckedIOException(ex);
                }
            });
        } catch (IOException e) {
            e.printStackTrace();
        }
 
        // dump memory data to file for testing
        Files.write(Paths.get("./users-memory.parquet"), outputFile.toArray());
    }

内存中的测试数据

$ parquet-tools cat --json users-memory.parquet
$ parquet-tools schema users-memory.parquet

【讨论】：

【解决方案2】：

"but it will write the Parquet file to the disk"
"if there was an easier way to just keep the file in memory"

根据您的查询，我了解到您不想将部分文件写入镶木地板。如果您希望将完整的文件以 parquet 格式写入磁盘并在内存中写入临时文件，您可以使用内存映射文件和 parquet 格式的组合。

将您的数据写入内存映射文件，完成写入后将字节转换为 parquet 格式并存储到磁盘。

看看MappedByteBuffer。

【讨论】：