[Java] Memory leak when use VectorSchemaSlice slice
- Lenguaje dominante
- Java
- Estrellas
- 94
- Forks
- 152
- Merge medio
- 3 d 16 h
- PR fusionados (30 d)
- 11
Descripción
### Describe the usage question you have. Please include as many useful details as possible.
In my Flink process function, I receive serialized VectorSchemaRoot data which needs to be deserialized for further processing. As I need to operate on it row by row, I utilized the slice function. However, this approach can lead to an increase in direct memory usage which in turn reduces the heap memory space in Flink. Ultimately, this can result in an Out Of Memory (OOM) exception as the heap memory space becomes insufficient. However, if I serialize the sliced VectorSchemaRoot and pass it as a parameter to the downstream function, which then deserializes it again, there will be no more OOM issues.
```
public void processElement(I value, ProcessFunction.Context context, Collector collector) throws Exception {
if (this.writerHelper == null) {
initWriterHelper();
}
reader = new ArrowStreamReader(new ByteArrayInputStream((byte[]) value), ArrowUtil.rootAllocator);
try {
while (reader.loadNextBatch()) {
VectorSchemaRoot vsr = reader.getVectorSchemaRoot();
int rowCount = vsr.getRowCount();
for (int i = 0; i < rowCount; i++) {
//split to row
VectorSchemaRoot row = vsr.slice(i, 1);
// this.writerHelper.write(row); This approach can result in a memory leak, whereas the following method will not.
ByteArrayOutputStream out = new ByteArrayOutputStream();
ArrowStreamWriter writer =
new ArrowStreamWriter(row, null, Channels.newChannel(out));
writer.start();
writer.writeBatch();
this.writerHelper.write(out.toByteArray());
row.clear();
row.close();
}
vsr.clear();
vsr.close();
}
} catch (Exception ex) {
ex.printStackTrace();
} finally {
reader.close();
}
}```
### Component(s)
Java
Guía de contribución
Línea de trabajo
Empieza por el ejemplo de processElement, especialmente por VectorSchemaRoot.slice(i, 1) y las llamadas row.clear()/row.close(). Reproduce el crecimiento de la memoria directa al escribir filas segmentadas mediante writerHelper y compáralo después con la ruta de serialización y deserialización. Se considera terminado cuando la ruta de filas segmentadas ya no provoca el crecimiento de memoria reportado ni OOM.
Escrito por el modelo de indexación a partir del texto del issue.
Evaluación
- Stack tecnológico
- java
- Área
- data
- Tipo de issue
- Error
- Dificultad
- 4/5
- Tiempo estimado
- 3-5 días
- Estado de actividad
- Estancado
- Claridad
- Necesita aclaración
- Aptitud para principiantes
- 30/100