JVM上高性能数据格式库包Apache Arrow入门和架构详解(Gkatziouras)
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入门的资料请关注其它相关文章!