我有一个 java 应用程序,可以将 json 消息转换为 parquet 格式。 java中是否有任何镶木地板写入器可以写入缓冲区或字节流?大多数示例,我都见过写入文件。
我还需要写入流,所以我完成了 naimdjon 给出的示例。以下对我来说非常有效。
class ParquetBufferedWriter implements OutputFile {
private final BufferedOutputStream out;
public ParquetBufferedWriter(BufferedOutputStream out) {
this.out = out;
}
@Override
public PositionOutputStream create(long blockSizeHint) throws IOException {
return createPositionOutputstream();
}
private PositionOutputStream createPositionOutputstream() {
return new PositionOutputStream() {
int pos = 0;
@Override
public long getPos() throws IOException {
return pos;
}
@Override
public void flush() throws IOException {
out.flush();
};
@Override
public void close() throws IOException {
out.close();
};
@Override
public void write(int b) throws IOException {
out.write(b);
pos++;
}
@Override
public void write(byte[] b, int off, int len) throws IOException {
out.write(b, off, len);
pos += len;
}
};
}
@Override
public PositionOutputStream createOrOverwrite(long blockSizeHint) throws IOException {
return createPositionOutputstream();
}
@Override
public boolean supportsBlockSize() {
return false;
}
@Override
public long defaultBlockSize() {
return 0;
}
}
TLDR;您需要实施
OutputFile
,例如类似的东西:
import org.apache.parquet.io.OutputFile;
import org.apache.parquet.io.PositionOutputStream;
import java.io.BufferedOutputStream;
import java.io.IOException;
public class ParquetBufferedWriter implements OutputFile {
private final BufferedOutputStream out;
public ParquetBufferedWriter(BufferedOutputStream out) {
this.out = out;
}
@Override
public PositionOutputStream create(long blockSizeHint) throws IOException {
return createPositionOutputstream();
}
private PositionOutputStream createPositionOutputstream() {
int pos = 0;
return new PositionOutputStream() {
@Override
public long getPos() throws IOException {
return pos;
}
@Override
public void write(int b) throws IOException {
out.write(b);
pos++;
}
@Override
public void write(byte[] b, int off, int len) throws IOException {
out.write(b, off, len);
pos += len;
}
};
}
@Override
public PositionOutputStream createOrOverwrite(long blockSizeHint) throws IOException {
return createPositionOutputstream();
}
@Override
public boolean supportsBlockSize() {
return false;
}
@Override
public long defaultBlockSize() {
return 0;
}
}
你的作家会是这样的:
ParquetBufferedWriter out = new ParquetBufferedWriter();
try (ParquetWriter<Record> writer = AvroParquetWriter.
<Record>builder(out)
.withRowGroupSize(DEFAULT_BLOCK_SIZE)
.withPageSize(DEFAULT_PAGE_SIZE)
.withSchema(SCHEMA)
.build()) {
for (Record record : records) {
writer.write(record);
}
} catch (IOException e) {
throw new IllegalStateException(e);
}
您需要将数据写入临时文件,然后将数据从文件转换到输入流或缓冲区 像这样,首先读取临时文件数据
final InputStream targetStream = new DataInputStream(new FileInputStream(tmp1.getAbsoluteFile()));
StringWriter writer = new StringWriter();
String encoding = StandardCharsets.UTF_8.name();
IOUtils.copy(targetStream, writer, encoding);
System.out.println(writer);