JVM上高性能数据格式库包Apache Arrow入门和架构详解(Gkatziouras)

作者: 保持微笑e
来源: 51数据库
2021-06-24

apache arrow是是各种大数据工具（包括bigquery）使用的一种流行格式，它是平面和分层数据的存储格式。它是一种加快应用程序内存密集型。

数据处理和数据科学领域中的常用库： apache arrow 。诸如apache parquet，apache spark，pandas之类的开放源代码项目以及许多商业或封闭源代码服务都使用arrow。它提供以下功能：

内存计算
标准化的柱状存储格式
一个ipc和rpc框架，分别用于进程和节点之间的数据交换

让我们看一看在arrow出现之前事物是如何工作的：

我们可以看到，为了使spark从parquet文件中读取数据，我们需要以parquet格式读取和反序列化数据。这要求我们通过将数据加载到内存中来制作数据的完整副本。首先，我们将数据读入内存缓冲区，然后使用parquet的转换方法将数据（例如字符串或数字）转换为我们的编程语言的表示形式。这是必需的，因为parquet表示的数字与python编程语言表示的数字不同。

由于许多原因，这对于性能来说是一个很大的问题：

我们正在复制数据并在其上运行转换步骤。数据的格式不同，我们需要对所有数据进行读取和转换，然后再对数据进行任何计算。
我们正在加载的数据必须放入内存中。您只有8gb的ram，数据是10gb吗？你真倒霉！

现在，让我们看一下apache arrow如何改进这一点：

arrow无需复制和转换数据，而是了解如何直接读取和操作数据。为此，arrow社区定义了一种新的文件格式以及直接对序列化数据起作用的操作。可以直接从磁盘读取此数据格式，而无需将其加载到内存中并转换/反序列化数据。当然，部分数据仍将被加载到ram中，但您的数据不必放入内存中。arrow使用其文件的内存映射功能，仅在必要和可能的情况下将尽可能多的数据加载到内存中。

apache arrow支持以下语言：

c++
c#
go
java
javascript
rust
python (through the c++ library)
ruby (through the c++ library)
r (through the c++ library)
matlab (through the c++ library).

arrow特点

arrow首先是提供用于内存计算的列式数据结构的库，可以将任何数据解压缩并解码为arrow柱状数据结构，以便随后可以对解码后的数据进行内存内分析。arrow列格式具有一些不错的属性：随机访问为o（1），每个值单元格在内存中的前一个和后一个相邻，因此进行迭代非常有效。

apache arrow定义了一种二进制“序列化”协议，用于安排arrow列数组的集合（称为“记录批处理”），该数组可用于消息传递和进程间通信。您可以将协议放在任何地方，包括磁盘上，以后可以对其进行内存映射或读入内存并发送到其他地方。

arrow协议的设计目的是使您可以“映射”一个arrow数据块而不进行任何反序列化，因此对磁盘上的arrow协议数据执行分析可以使用内存映射并有效地支付零成本。该协议用于很多事情，例如spark sql和python之间的流数据，用于针对spark sql数据块运行pandas函数，这些被称为“ pandas udfs”。

arrow是为内存而设计的（但是您可以将其放在磁盘上，然后再进行内存映射）。它们旨在相互兼容，并在应用程序中一起使用，而其竞争对手apache parquet文件是为磁盘存储而设计的。

优点：apache arrow为平面和分层数据定义了一种独立于语言的列式存储格式，该格式组织为在cpu和gpu等现代硬件上进行高效的分析操作而组织。arrow存储器格式还支持零拷贝读取，以实现闪电般的数据访问，而无需序列化开销。

java的apache arrow

导入库：

<dependency>
    <groupid>org.apache.arrow</groupid>
    <artifactid>arrow-memory-netty</artifactid>
    <version>${arrow.version}</version>
</dependency>
<dependency>
    <groupid>org.apache.arrow</groupid>
    <artifactid>arrow-vector</artifactid>
    <version>${arrow.version}</version>
</dependency>

在开始之前，必须了解对于arrow的读/写操作，使用了字节缓冲区。诸如读取和写入之类的操作是字节的连续交换。为了提高效率，arrow附带了一个缓冲区分配器，该缓冲区分配器可以具有一定的大小，也可以具有自动扩展功能。支持分配管理的库是arrow-memory-netty和arrow-memory-unsafe。我们这里使用netty。

用arrow存储数据需要一个模式，模式可以通过编程定义：

package com.gkatzioura.arrow;

import java.io.ioexception;

import java.util.list;

import org.apache.arrow.vector.types.pojo.arrowtype;

import org.apache.arrow.vector.types.pojo.field;

import org.apache.arrow.vector.types.pojo.fieldtype;

import org.apache.arrow.vector.types.pojo.schema;

public class schemafactory {

public static schema default_schema = createdefault();

public static schema createdefault() {

var strfield = new field("col1", fieldtype.nullable(new arrowtype.utf8()), null);

var intfield = new field("col2", fieldtype.nullable(new arrowtype.int(32, true)), null);

return new schema(list.of(strfield, intfield));

}

public static schema schemawithchildren() {

var amount = new field("amount", fieldtype.nullable(new arrowtype.decimal(19,4,128)), null);

var currency = new field("currency",fieldtype.nullable(new arrowtype.utf8()), null);

var itemfield = new field("item", fieldtype.nullable(new arrowtype.utf8()), list.of(amount,currency));

return new schema(list.of(itemfield));

}

public static schema fromjson(string jsonstring) {

try {

return schema.fromjson(jsonstring);

} catch (ioexception e) {

throw new arrowexampleexception(e);

}

}

}

他们也有一个可解析的json表示形式：

{
  "fields" : [ {
    "name" : "col1",
    "nullable" : true,
    "type" : {
      "name" : "utf8"
    },
    "children" : [ ]
  }, {
    "name" : "col2",
    "nullable" : true,
    "type" : {
      "name" : "int",
      "bitwidth" : 32,
      "issigned" : true
    },
    "children" : [ ]
  } ]
}

另外，就像avro一样，您可以在字段上设计复杂的架构和嵌入式值：

public static schema schemawithchildren() {
    var amount = new field("amount", fieldtype.nullable(new arrowtype.decimal(19,4,128)), null);
    var currency = new field("currency",fieldtype.nullable(new arrowtype.utf8()), null);
    var itemfield = new field("item", fieldtype.nullable(new arrowtype.utf8()), list.of(amount,currency));
 
    return new schema(list.of(itemfield));
}

基于上面的的schema，我们将为我们的类创建一个dto：

package com.gkatzioura.arrow;
 
import lombok.builder;
import lombok.data;
 
@data
@builder
public class defaultarrowentry {
 
    private string col1;
    private integer col2;
 
}

我们的目标是将这些java对象转换为arrow字节流。

1. 使用分配器创建 directbytebuffer

这些缓冲区是的。您确实需要释放所使用的内存，但是对于库用户而言，这是通过在分配器上执行 close（）操作来完成的。在我们的例子中，我们的类将实现 closeable 接口，该接口将执行分配器关闭操作。

通过使用流api，数据将被流传输到使用arrow格式提交的outputstream：

package com.gkatzioura.arrow;
 
import java.io.closeable;
import java.io.ioexception;
import java.nio.channels.writablebytechannel;
import java.util.list;
 
import org.apache.arrow.memory.rootallocator;
import org.apache.arrow.vector.intvector;
import org.apache.arrow.vector.varcharvector;
import org.apache.arrow.vector.vectorschemaroot;
import org.apache.arrow.vector.dictionary.dictionaryprovider;
import org.apache.arrow.vector.ipc.arrowstreamwriter;
import org.apache.arrow.vector.util.text;
 
import static com.gkatzioura.arrow.schemafactory.default_schema;
 
public class defaultentrieswriter implements closeable {
 
    private final rootallocator rootallocator;
    private final vectorschemaroot vectorschemaroot;//向量分配器创建：
 
    public defaultentrieswriter() {
        rootallocator = new rootallocator();
        vectorschemaroot = vectorschemaroot.create(default_schema, rootallocator);
    }
 
    public void write(list<defaultarrowentry> defaultarrowentries, int batchsize, writablebytechannel out) {
        if (batchsize <= 0) {
            batchsize = defaultarrowentries.size();
        }
 
        dictionaryprovider.mapdictionaryprovider dictprovider = new dictionaryprovider.mapdictionaryprovider();
        try(arrowstreamwriter writer = new arrowstreamwriter(vectorschemaroot, dictprovider, out)) {
            writer.start();
 
            varcharvector childvector1 = (varcharvector) vectorschemaroot.getvector(0);
            intvector childvector2 = (intvector) vectorschemaroot.getvector(1);
            childvector1.reset();
            childvector2.reset();
 
            boolean exactbatches = defaultarrowentries.size()%batchsize == 0;
            int batchcounter = 0;
 
            for(int i=0; i < defaultarrowentries.size(); i++) {
                childvector1.setsafe(batchcounter, new text(defaultarrowentries.get(i).getcol1()));
                childvector2.setsafe(batchcounter, defaultarrowentries.get(i).getcol2());
 
                batchcounter++;
 
                if(batchcounter == batchsize) {
                    vectorschemaroot.setrowcount(batchsize);
                    writer.writebatch();
                    batchcounter = 0;
                }
            }
 
            if(!exactbatches) {
                vectorschemaroot.setrowcount(batchcounter);
                writer.writebatch();
            }
 
            writer.end();
        } catch (ioexception e) {
            throw new arrowexampleexception(e);
        }
    }
 
    @override
    public void close() throws ioexception {
        vectorschemaroot.close();
        rootallocator.close();
    }
 
}

为了在arrow上显示批处理的支持，已在函数中实现了简单的批处理算法。对于我们的示例，只需考虑将数据分批写入。

让我们深入了解上面代码功能：

向量分配器创建：

public defaultentriestobytesconverter() {
    rootallocator = new rootallocator();
    vectorschemaroot = vectorschemaroot.create(default_schema, rootallocator);
}

然后在写入流时，实现并启动了arrow流编写器

arrowstreamwriter writer = new arrowstreamwriter(vectorschemaroot, dictprovider, channels.newchannel(out));
writer.start();

我们将数据填充向量，然后还重置它们，但让预分配的缓冲区存在：

varcharvector childvector1 = (varcharvector) vectorschemaroot.getvector(0);
intvector childvector2 = (intvector) vectorschemaroot.getvector(1);
childvector1.reset();
childvector2.reset();

写入数据时，我们使用 setsafe 操作。如果需要分配更多的缓冲区，应采用这种方式。对于此示例，此操作在每次写入时都完成，但是在考虑了所需的操作和缓冲区大小后可以避免：

childvector1.setsafe(i, new text(defaultarrowentries.get(i).getcol1()));
childvector2.setsafe(i, defaultarrowentries.get(i).getcol2());

然后，将批处理写入流中：

vectorschemaroot.setrowcount(batchsize);
writer.writebatch();

最后但并非最不重要的一点是，我们关闭了writer:

@override
public void close() throws ioexception {
    vectorschemaroot.close();
    rootallocator.close();
}

以上就是jvm上高性能数据格式库包apache arrow入门和架构详解(gkatziouras)的详细内容，更多关于apache arrow入门的资料请关注其它相关文章！