欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  网络运营

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

程序员文章站 2022-03-15 21:22:04
apache arrow是是各种大数据工具(包括bigquery)使用的一种流行格式,它是平面和分层数据的存储格式。它是一种加快应用程序内存密集型。数据处理和数据科学领域中的常用库: apache a...

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

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

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

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

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

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

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

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

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

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

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入门的资料请关注其它相关文章!