标签 Cpp 下的文章

Go语言开发者的Apache Arrow使用指南:扩展compute包

本文永久链接 – https://tonybai.com/2023/07/22/a-guide-of-using-apache-arrow-for-gopher-part5

在本系列文章的第4篇《Go语言开发者的Apache Arrow使用指南:数据操作》中我们遇到了大麻烦:Go的Arrow实现居然不支持像max、min、sum这样的简单聚合计算函数:(,分组聚合(grouped aggregation)就更是“遥不可期”。要想对从CSV读取的数据进行聚合操作和分析,我们只能“自己动手,丰衣足食” – 扩展Arrow Go实现中的compute包了

不过,Arrow的Go实现还是蛮复杂的,如果对其结构没有一个初步的认知,很难实现这类扩展。在这篇文章中,我们就来了解一下compute包的结构,并尝试为compute包添加几个简单的、仅能处理单一类型的聚合函数,先来完成一些从0到1的工作。

为了深入了解Go Arrow实现,我又翻阅了一下Arrow官方的文档,显然Arrow C++的文档是最丰富的。我快读了一下C++的Arrow文档,对Arrow的结构有了更深刻的认知,基于这些资料,我们先来做一下Arrow结构的回顾。

0. 回顾Arrow的各个layer

Arrow的C++文档使用layer来介绍各种Arrow的概念,我们挑几个重要的看一下:

  • 物理层(The physical layer)

物理层针对的是内存的分配管理,包括内存分配的方法(堆分配器、内存文件映射、静态内存区)等。这一层的一个最重要的概念就是我们之前在数据类型一文中提到的Buffer抽象,它代表了内存中的一块连续的数据存储区域

  • 一维表示层(The one-dimensional layer)

除了物理层,后续的层都是逻辑层。一维表示层是一个逻辑表示层,它定义了Arrow的最基本数据类型:array数据类型决定了物理层内存数据的解释方法,逻辑数据类型array在物理层投影为一个和多个内存buffer

我们在“高级数据结构”提到的chunked array也在这一层,chunked array由多个同构类型的array组成,Arrow将其理解为一个同构的(相同类型的)、逻辑上值连续的、更大的array,是array基础类型的一个更泛化的表示。

  • 二维表示层(The two-dimensional layer)

“高级数据结构”一文中除chunked array之外的概念,都在这一层,包括schema、table、record batch。

schema是用于描述一维数据(一列数据,即一个逻辑array)的元数据,包括列名、类型与其他元信息。

Table是schema+与schema元信息对应的多个chunked array,它是Arrow中数据集抽象能力最强的逻辑结构。

Record Batch则是schema+与schema元信息对应的多个array。还记得“高级数据结构”一文中的那副直观给出table与record batch差异的图么:

  • 计算层(The compute layer)

计算层一个重要的抽象是Datum,这是一个灵活的抽象,用于统一表示参与计算的各类输入参数和返回值。

计算层真正执行计算的函数被统一放在kernel这个“层次”中,这个层次的函数对Datum类型的输入参数进行计算并返回Datam类型的结果或以Datum类型的输出参数承载计算结果。

  • IPC层(The Inter-Process Communication (IPC) layer)

这是我们尚未接触过的一层,通过这一层,复合Arrow columnar format的数据可以在进程间(同一主机或不同主机)交互,并且这种交换可以保证尽可能少的内存copy。

  • 文件格式层(The file formats layer)

这一层负责读写文件,在之前的“数据操作”一篇中,我们接触过将CSV文件中的数据读到内存中并组织为Arrow列式存储格式,在后续篇章中,我们还将陆续介绍Arrow与CSV(写入)、Parquet文件的数据交互。

C++有关Arrow的介绍中还有设备层(the devices layer)、文件系统层(the file system layer)等,后续可能不会涉及,这里就不说了。

通过上述回顾,再对照本系列第一篇文章“数据类型”的内容,你对Arrow的理解是不是更深刻一点点了呢:)。

接下来,我们重点看看计算层(the compute layer)。

1. 计算层(the compute layer)的结构

Go语言的计算层在compute目录下。Go语言借鉴了C++计算层的设计,将计算层分为compute和kernel,这个从代码布局上也可以明显看出来:

$tree -F -L 2 compute|grep -v go
compute           --- compute层
├── exprs/
├── internal/
│   ├── exec/
│   └── kernels/  --- compute的kernel层

compute包采用了registry模式,初始化时将底层的kernel function包装成上层的Function并注册到registry中。用户调用某个function时,该function会在registry中查找对应的注册函数并调用。

下面我们通过Uniq这个array-wise函数作为例子来探索一下kernel function的注册与调用过程。下面是“数据操作”一文中的示例,这里再次借用一下:

// arrow/manipulation/unary_arraywise_function.go

func main() {
    data := []int32{5, 10, 0, 25, 2, 10, 2, 25}
    bldr := array.NewInt32Builder(memory.DefaultAllocator)
    defer bldr.Release()
    bldr.AppendValues(data, nil)
    arr := bldr.NewArray()
    defer arr.Release()

    dat, err := compute.Unique(context.Background(), compute.NewDatum(arr))
    if err != nil {
        fmt.Println(err)
        return
    }

    arr1, ok := dat.(*compute.ArrayDatum)
    if !ok {
        fmt.Println("type assert fail")
        return
    }
    fmt.Println(arr1.MakeArray()) // [5 10 0 25 2]
}

下面是Unique函数的注册和调用过程示意图:

很显然,整个过程包括两个明显的阶段:

  • 包装并向Registry注册kernel函数(AddFunction)
  • 在Registry中查找函数并调用(GetFunction)

当我们在用户层调用compute.Unique函数时,一个统一的CallFunction会被调用,其第二个参数”uniq”表明我们要调用registry中的名为”uniq”的包装函数。在这个过程中GetFunctionRegistry被调用以获取registry实例,在这个过程中,如果registry实例尚没有创建,GetFunctionRegistry会在sync.Once的保护下创建registry并进行初始注册工作(RegisterXXX)。”uniq”对应的包装函数是在RegisterVectorHash中被注册到registry中的。

RegisterVectorHash会通过kernel层提供的GetVectorHashKernels获取kernel层的”uniq”实现,并将其通过NewVectorFunction和AddKernel包装为uniqFn这一用户层的Function,该uniqFn Function最终会被AddFunction加入到registry中。

而CallFunction(ctx, “uniq”)也会从registry中将uniqFn查找出来并执行其Execute方法,该Execute方法实际上执行的是kernel层的”uniq”实现。

我们看到:通过示意图展示的Unique函数的注册与调用过程还是相对清晰的(但如果要阅读对应的代码,还是比较繁琐的)。

到这里我们也大致了解了compute包的结构以及与kernel层的关系,接下来我们就来尝试给compute包添加一些scalar aggregate函数,所谓scalar aggregate函数就是输入是array,输出是一个scalar值的函数,比如:max、min、sum等。

3. 添加Max、Min、Sum、Avg等Scalar Aggregate函数

在上一篇“数据操作”时提过,聚合函数分为Scalar聚合和grouped聚合,显然Scalar聚合函数要简单一些,这里我们就来向compute层添加scalar aggregate函数,以Max为例,我们希望用户层这样使用Max聚合函数:

// max_aggregate_function.go
func main() {
    data := []int64{5, 10, 0, 25, 2, 35, 7, 15}
    bldr := array.NewInt64Builder(memory.DefaultAllocator)
    defer bldr.Release()
    bldr.AppendValues(data, nil)
    arr := bldr.NewArray()
    defer arr.Release()

    dat, err := compute.Max(context.Background(), compute.NewDatum(arr))
    if err != nil {
        fmt.Println(err)
        return
    }

    ad, ok := dat.(*compute.ArrayDatum)
    if !ok {
        fmt.Println("type assert fail")
        return
    }
    arr1 := ad.MakeArray()
    fmt.Println(arr1) // [35]
}

注:这里有一个问题,那就是Max返回的Datum是一个ArrayDatum,而不是期望的ScalarDatum。

通过上面的compute layer的结构,我们知道,如果要添加Max、Min、Sum、Avg等Scalar Aggregate函数,我们需要在kernel层和compute层协作实现。下面是实现的具体步骤。

3.1 向kernel层添加scalar聚合函数实现

compute层要支持scalar聚合,需要kernel层线支持scalar聚合,这里我们先向compute/internal/kernels目录添加一个scalar_agg.go,用于在kernel层实现scalar聚合,以Max为例:

// compute/internal/kernels/scalar_agg.go

package kernels

import (
    "fmt"

    "github.com/apache/arrow/go/v13/arrow"
    "github.com/apache/arrow/go/v13/arrow/compute/internal/exec"
    "github.com/apache/arrow/go/v13/arrow/scalar"
)

func ScalarAggKernels(op ScalarAggOperator) (aggs []exec.ScalarKernel) {
    switch op {
    case AggMax:
        maxAggs := maxAggKernels()
        aggs = append(aggs, maxAggs...)
    case AggMin:
        minAggs := minAggKernels()
        aggs = append(aggs, minAggs...)
    case AggAvg:
        avgAggs := avgAggKernels()
        aggs = append(aggs, avgAggs...)
    case AggSum:
        sumAggs := sumAggKernels()
        aggs = append(aggs, sumAggs...)
    }

    return
}

func aggMax(ctx *exec.KernelCtx, batch *exec.ExecSpan, out *exec.ExecResult) error {
    var max int64

    for _, v := range batch.Values {
        if !v.IsArray() {
            return fmt.Errorf("%w: input datum is not array", arrow.ErrInvalid)
        }

        if v.Array.Type != arrow.PrimitiveTypes.Int64 {
            return fmt.Errorf("%w: array type is not int64", arrow.ErrInvalid)
        }

        // for int64 array:
        //   first buffer is meta buffer
        //   second buffer is what we want
        int64s := exec.GetSpanValues[int64](&v.Array, 1)
        for _, v64 := range int64s {
            if v64 > max {
                max = v64
            }
        }
    }

    out.FillFromScalar(scalar.NewInt64Scalar(max))
    return nil
}

func maxAggKernels() (aggs []exec.ScalarKernel) {
    outType := exec.NewOutputType(arrow.PrimitiveTypes.Int64)
    in := exec.NewExactInput(arrow.PrimitiveTypes.Int64)
    aggs = append(aggs, exec.NewScalarKernel([]exec.InputType{in}, outType,
        aggMax, nil))

    return
}
... ...

上面的ScalarAggKernels函数就像上图中的GetVectorHashKernels一样,为compute层提供kernel层scalar agg函数的获取“渠道”。aggMax函数是实现聚合逻辑的那个函数,它针对输入的array进行操作,计算array中所有元素中的最大值,并将这个值包装成Datum作为out参数输出。

在compute/internal/kernels/types.go中,我们定义了如下枚举常量,用于compute层传入要选择的scalar聚合函数。

// compute/internal/kernels/types.go

//go:generate stringer -type=ScalarAggOperator -linecomment

type ScalarAggOperator int8

const (
    AggMax ScalarAggOperator = iota // max
    AggMin                          // min
    AggAvg                          // avg
    AggSum                          // sum
)

3.2 在compute层提供对kernel层聚合函数的包装

在compute层,我们也提供一个scalar_agg.go文件,用于对kernel层的聚合函数进行包装:

// compute/scalar_agg.go

package compute

import (
    "context"

    "github.com/apache/arrow/go/v13/arrow/compute/internal/kernels"
)

type aggFunction struct {
    ScalarFunction
}

func Max(ctx context.Context, values Datum) (Datum, error) {
    return CallFunction(ctx, "max", nil, values)
}
func Min(ctx context.Context, values Datum) (Datum, error) {
    return CallFunction(ctx, "min", nil, values)
}
func Avg(ctx context.Context, values Datum) (Datum, error) {
    return CallFunction(ctx, "avg", nil, values)
}
func Sum(ctx context.Context, values Datum) (Datum, error) {
    return CallFunction(ctx, "sum", nil, values)
}

func RegisterScalarAggs(reg FunctionRegistry) {
    maxFn := &aggFunction{*NewScalarFunction("max", Unary(), EmptyFuncDoc)}
    for _, k := range kernels.ScalarAggKernels(kernels.AggMax) {
        if err := maxFn.AddKernel(k); err != nil {
            panic(err)
        }
    }
    reg.AddFunction(maxFn, false)

    minFn := &aggFunction{*NewScalarFunction("min", Unary(), EmptyFuncDoc)}
    for _, k := range kernels.ScalarAggKernels(kernels.AggMin) {
        if err := minFn.AddKernel(k); err != nil {
            panic(err)
        }
    }
    reg.AddFunction(minFn, false)

    avgFn := &aggFunction{*NewScalarFunction("avg", Unary(), EmptyFuncDoc)}
    for _, k := range kernels.ScalarAggKernels(kernels.AggAvg) {
        if err := avgFn.AddKernel(k); err != nil {
            panic(err)
        }
    }
    reg.AddFunction(avgFn, false)

    sumFn := &aggFunction{*NewScalarFunction("sum", Unary(), EmptyFuncDoc)}
    for _, k := range kernels.ScalarAggKernels(kernels.AggSum) {
        if err := sumFn.AddKernel(k); err != nil {
            panic(err)
        }
    }
    reg.AddFunction(sumFn, false)
}

我们看到在这个源文件中,我们提供了供最终用户调用的Max等函数,这些函数是对kernel层scalar聚合函数的包装,通过CallFunction在registry中找到注册的kernel函数并执行它。

RegisterScalarAggs是用于向registry注册scalar聚合函数的函数。

3.3 在compute层将包装后的聚合函数注册到Registry中

我们修改一下compute/registry.go,在GetFunctionRegistry函数中增加对RegisterScalarAggs的调用,以实现对scalar聚合函数的注册:

// compute/registry.go

func GetFunctionRegistry() FunctionRegistry {
    once.Do(func() {
        registry = NewRegistry()
        RegisterScalarCast(registry)
        RegisterVectorSelection(registry)
        RegisterScalarBoolean(registry)
        RegisterScalarArithmetic(registry)
        RegisterScalarComparisons(registry)
        RegisterVectorHash(registry)
        RegisterVectorRunEndFuncs(registry)
        RegisterScalarAggs(registry)
    })
    return registry
}

3.4 运行示例

最初运行arrow/compute-extension/max_aggregate_function.go示例的结果并非我们预期,而是一个全0的数组:

$go run max_aggregate_function.go
[0 0 0 0 0 0 0 0]

经过print调试大法后,我发现compute/executor.go中的executeSpans的实现似乎有一个问题,我在arrow项目提了一个issue,并对executor.go做了如下修改:

diff --git a/go/arrow/compute/executor.go b/go/arrow/compute/executor.go
index d3f1a1fd4..e9bda7137 100644
--- a/go/arrow/compute/executor.go
+++ b/go/arrow/compute/executor.go
@@ -604,7 +604,7 @@ func (s *scalarExecutor) executeSpans(data chan<- Datum) (err error) {
                        return
                }

-               return s.emitResult(prealloc, data)
+               return s.emitResult(&output, data)
        }

        // fully preallocating, but not contiguously
(END)

修改后,再运行arrow/compute-extension/max_aggregate_function.go示例就得到了正确的结果:

$go run max_aggregate_function.go
[35]

3.5 To Be Done

到这里,我们从0到1的为arrow go实现的compute层添加了int64类型的scalar聚合函数的支持(以max为例),但这仅仅是验证了思路的可行性,上述对compute的修改可能是不合理的。此外,上述的改动不是production ready的,存在一些问题,比如:

  • Max返回的是array datam,而不是我们想要的scalar Datam;
  • 仅支持int64,不支持其他类型的max聚合,比如float64、string等;
  • 性能没有优化;
  • 对chunked array类型的scalar datam尚未给出验证示例。
  • … …

4. 小结

在本文中我们基于C++的资料,回顾了Arrow的一些基础抽象概念,从而对Arrow有了更为深刻的认知。之后,也是我们的重点,就是给出了compute层的结构以及基于该结构为compute层增加scalar聚合函数的一种思路和示例代码。

不过这种思路只是为了理解arrow的一种试验性方法,存在其不合理的地方,随着arrow演进,这种方法也许将不适用。同时,后续arrow官方可能会为go增加aggregate function的支持,那时请大家以官方实现为准。

C++版本Arrow实现完全支持各种聚合函数,考虑到Go arrow的实现参考了C++版本的思路,如果要为go arrow正式增加聚合函数支持,阅读c++源码并考虑迁移到Go才是正道。

本文示例代码可以在这里下载,同时增加了scalar function的arrow的fork版本可以在我的github项目arrow-extend-compute1下找到。

5. 参考资料

  • 计算层 – https://arrow.apache.org/docs/cpp/compute.html
  • 计算层教程 – https://arrow.apache.org/docs/cpp/tutorials/compute_tutorial.html
  • Arrow C++参考 – https://arrow.apache.org/docs/cpp/overview.html
  • Go unique kernel函数PR – https://github.com/apache/arrow/pull/34172

“Gopher部落”知识星球旨在打造一个精品Go学习和进阶社群!高品质首发Go技术文章,“三天”首发阅读权,每年两期Go语言发展现状分析,每天提前1小时阅读到新鲜的Gopher日报,网课、技术专栏、图书内容前瞻,六小时内必答保证等满足你关于Go语言生态的所有需求!2023年,Gopher部落将进一步聚焦于如何编写雅、地道、可读、可测试的Go代码,关注代码质量并深入理解Go核心技术,并继续加强与星友的互动。欢迎大家加入!

img{512x368}
img{512x368}

img{512x368}
img{512x368}

著名云主机服务厂商DigitalOcean发布最新的主机计划,入门级Droplet配置升级为:1 core CPU、1G内存、25G高速SSD,价格5$/月。有使用DigitalOcean需求的朋友,可以打开这个链接地址:https://m.do.co/c/bff6eed92687 开启你的DO主机之路。

Gopher Daily(Gopher每日新闻)归档仓库 – https://github.com/bigwhite/gopherdaily

我的联系方式:

  • 微博(暂不可用):https://weibo.com/bigwhite20xx
  • 微博2:https://weibo.com/u/6484441286
  • 博客:tonybai.com
  • github: https://github.com/bigwhite

商务合作方式:撰稿、出书、培训、在线课程、合伙创业、咨询、广告合作。

Go语言开发者的Apache Arrow使用指南:数据操作

本文永久链接 – https://tonybai.com/2023/07/13/a-guide-of-using-apache-arrow-for-gopher-part4

在前面的Arrow系列文章中,我们介绍了Arrow的基础数据类型以及高级数据类型,这让我们具备了在内存中建立起一个immutable数据集的能力。但这并非我们的目标,我们最终是要对建立起来的数据集进行查询和分析等操作(manipulation)的。

在这一篇文章中,我们就来看看如何基于Go arrow的实现对内存中的Arrow数据集进行操作。

注:由于Arrow官方文档尚没有Go语言的cookbook,这里的一些例子参考了其他语言的Cookbook,比如Python

1. 从CSV文件中读取数据

在操作数据之前,我们首先需要准备数据,并将数据读取到内存中以Arrow的列式存储形式组织起来。Arrow的Go实现支持从多种文件格式中将数据读取出来并在内存中构建出Arrow列式数据集:

从图中我们看到:Arrow Go支持读取的文件格式包括CSV、JSON和Parquet等。CSV、JSON都是日常最常用的文件格式,那么Parquet是什么呢?这是一种面向列的文件存储格式,支持高效的数据存储和数据获取能力。influxdb iox存储引擎采用的就是Apache Arrow + Parquet的经典组合。我们在本系列的后续文章中会单独说一下Arrow + Parquet,在本文中Parquet不是重点。

注:Parquet的读音是:[’pɑːkeɪ] 。

在这篇文章中,我们以从CSV文件中读取数据为例。我们的CSV文件来自于Kaggle平台上的开放数据集,这是一份记录着Delhi这个地方(应该是印度城市德里)1996年到2017年小时级的天气数据的CSV文件:testset.csv。该文件带有列头,有20列,10w多行记录。

我们先来小试牛刀,即取该csv文件前10几行,存成名为testset.tiny.csv的文件。我们编写一段Go程序来读取CSV中的数据并在内存中建立一个Arrow Record Batch!大家还记得Arrow Record Batch是什么结构了么?我们回顾一下“高级数据结构”中的那张图你就记起来了:

接下来我们就使用Arrow Go实现提供的csv包读取testset.tiny.csv文件并输出经由读出的数据建构的Record Batch:

// read_tiny_csv_multi_trunks.go

package main

import (
    "fmt"
    "io"
    "os"

    "github.com/apache/arrow/go/v13/arrow/csv"
)

func read(data io.ReadCloser) error {
    // read 5 lines at a time to create record batches
    rdr := csv.NewInferringReader(data, csv.WithChunk(5),
        // strings can be null, and these are the values
        // to consider as null
        csv.WithNullReader(true, "", "null", "[]"),
        // assume the first line is a header line which names the columns
        csv.WithHeader(true))

    for rdr.Next() {
        rec := rdr.Record()
        fmt.Println(rec)
    }

    return nil
}

func main() {
    data, err := os.Open("./testset.tiny.csv")
    if err != nil {
        panic(err)
    }
    read(data)
}

这里的csv包可不是标准库中的那个包,而是Arrow Go实现中专门用于将csv文件数据读取并转换为Arrow内存对象的包。csv包提供了两个创建csv.Reader实例的函数,这里使用的是NewInferringReader(即带列类型推导的Reader)。该函数可以自动读取位于第一行的csv文件的header,获取列字段的名称与个数,形成Record的schema,并在读取下一条记录时尝试推导(infer)这一列的类型(data type)。

这里在调用NewInferringReader时还传入了一个功能选项开关WithChunk(5),即一次读取5条记录来构建一个新的Record Batch。

我们运行一下上面的代码:

$go run read_tiny_csv_multi_trunks.go
record:
  schema:
  fields: 20
    - datetime_utc: type=utf8, nullable
    -  _conds: type=utf8, nullable
    -  _dewptm: type=int64, nullable
    -  _fog: type=int64, nullable
    -  _hail: type=int64, nullable
    -  _heatindexm: type=utf8, nullable
    -  _hum: type=int64, nullable
    -  _precipm: type=utf8, nullable
    -  _pressurem: type=int64, nullable
    -  _rain: type=int64, nullable
    -  _snow: type=int64, nullable
    -  _tempm: type=int64, nullable
    -  _thunder: type=int64, nullable
    -  _tornado: type=int64, nullable
    -  _vism: type=int64, nullable
    -  _wdird: type=int64, nullable
    -  _wdire: type=utf8, nullable
    -  _wgustm: type=utf8, nullable
    -  _windchillm: type=utf8, nullable
    -  _wspdm: type=float64, nullable
  rows: 5
  col[0][datetime_utc]: ["19961101-11:00" "19961101-12:00" "19961101-13:00" "19961101-14:00" "19961101-16:00"]
  col[1][ _conds]: ["Smoke" "Smoke" "Smoke" "Smoke" "Smoke"]
  col[2][ _dewptm]: [9 10 11 10 11]
  col[3][ _fog]: [0 0 0 0 0]
  col[4][ _hail]: [0 0 0 0 0]
  col[5][ _heatindexm]: [(null) (null) (null) (null) (null)]
  col[6][ _hum]: [27 32 44 41 47]
  col[7][ _precipm]: [(null) (null) (null) (null) (null)]
  col[8][ _pressurem]: [1010 -9999 -9999 1010 1011]
  col[9][ _rain]: [0 0 0 0 0]
  col[10][ _snow]: [0 0 0 0 0]
  col[11][ _tempm]: [30 28 24 24 23]
  col[12][ _thunder]: [0 0 0 0 0]
  col[13][ _tornado]: [0 0 0 0 0]
  col[14][ _vism]: [5 (null) (null) 2 (null)]
  col[15][ _wdird]: [280 0 0 0 0]
  col[16][ _wdire]: ["West" "North" "North" "North" "North"]
  col[17][ _wgustm]: [(null) (null) (null) (null) (null)]
  col[18][ _windchillm]: [(null) (null) (null) (null) (null)]
  col[19][ _wspdm]: [7.4 (null) (null) (null) 0]

我们看到结果输出了将csv文件中数据读取并转换后的Record Batch的信息!

不过这个结果有一个问题,那就是我们的testset.tiny.csv有12行数据,上述结果为什么仅读出了5行呢?利用go.work引用本地下载的arrow代码做一下print调试后发现这样的一个错误:

strconv.ParseInt: parsing "1.2": invalid syntax

翻看一下testset.tiny.csv文件,在第五行发现了包含1.2的这条数据:

19961101-16:00,Smoke,11,0,0,,47,,1011,0,0,23,0,0,1.2,0,North,,,0

1.2这数据对应的是” _vism”这一列,我们看一下上面这一列的schema信息:

-  _vism: type=int64, nullable

我们看到NewInferringReader将这一列识别成int64类型了!NewInferringReader是根据第一行数据中来做类型推导的,而vism这一列的第一条数据恰为5,将其推导为int64也就不足为奇了。那么如何修正上述问题呢?NewInferringReader提供了一个WithColumnTypes的功能选项,通过它我们可以指定vism列的Arrow DataType:

    rdr := csv.NewInferringReader(data, csv.WithChunk(5),
        // strings can be null, and these are the values
        // to consider as null
        csv.WithNullReader(true, "", "null", "[]"),
        // assume the first line is a header line which names the columns
        csv.WithHeader(true),
        csv.WithColumnTypes(map[string]arrow.DataType{
            " _vism": arrow.PrimitiveTypes.Float64,
        }),
    )

修改后,我们再来运行一下read_tiny_csv_multi_trunks.go这个文件的代码:

$go run read_tiny_csv_multi_trunks.go
record:
  schema:
  fields: 20
    - datetime_utc: type=utf8, nullable
    -  _conds: type=utf8, nullable
    -  _dewptm: type=int64, nullable
    -  _fog: type=int64, nullable
    -  _hail: type=int64, nullable
    -  _heatindexm: type=utf8, nullable
    -  _hum: type=int64, nullable
    -  _precipm: type=utf8, nullable
    -  _pressurem: type=int64, nullable
    -  _rain: type=int64, nullable
    -  _snow: type=int64, nullable
    -  _tempm: type=int64, nullable
    -  _thunder: type=int64, nullable
    -  _tornado: type=int64, nullable
    -  _vism: type=float64, nullable
    -  _wdird: type=int64, nullable
    -  _wdire: type=utf8, nullable
    -  _wgustm: type=utf8, nullable
    -  _windchillm: type=utf8, nullable
    -  _wspdm: type=float64, nullable
  rows: 5
  col[0][datetime_utc]: ["19961101-11:00" "19961101-12:00" "19961101-13:00" "19961101-14:00" "19961101-16:00"]
  col[1][ _conds]: ["Smoke" "Smoke" "Smoke" "Smoke" "Smoke"]
  col[2][ _dewptm]: [9 10 11 10 11]
  col[3][ _fog]: [0 0 0 0 0]
  col[4][ _hail]: [0 0 0 0 0]
  col[5][ _heatindexm]: [(null) (null) (null) (null) (null)]
  col[6][ _hum]: [27 32 44 41 47]
  col[7][ _precipm]: [(null) (null) (null) (null) (null)]
  col[8][ _pressurem]: [1010 -9999 -9999 1010 1011]
  col[9][ _rain]: [0 0 0 0 0]
  col[10][ _snow]: [0 0 0 0 0]
  col[11][ _tempm]: [30 28 24 24 23]
  col[12][ _thunder]: [0 0 0 0 0]
  col[13][ _tornado]: [0 0 0 0 0]
  col[14][ _vism]: [5 (null) (null) 2 1.2]
  col[15][ _wdird]: [280 0 0 0 0]
  col[16][ _wdire]: ["West" "North" "North" "North" "North"]
  col[17][ _wgustm]: [(null) (null) (null) (null) (null)]
  col[18][ _windchillm]: [(null) (null) (null) (null) (null)]
  col[19][ _wspdm]: [7.4 (null) (null) (null) 0]

record:
  schema:
  fields: 20
    - datetime_utc: type=utf8, nullable
    -  _conds: type=utf8, nullable
    -  _dewptm: type=int64, nullable
    -  _fog: type=int64, nullable
    -  _hail: type=int64, nullable
    -  _heatindexm: type=utf8, nullable
    -  _hum: type=int64, nullable
    -  _precipm: type=utf8, nullable
    -  _pressurem: type=int64, nullable
    -  _rain: type=int64, nullable
    -  _snow: type=int64, nullable
    -  _tempm: type=int64, nullable
    -  _thunder: type=int64, nullable
    -  _tornado: type=int64, nullable
    -  _vism: type=float64, nullable
    -  _wdird: type=int64, nullable
    -  _wdire: type=utf8, nullable
    -  _wgustm: type=utf8, nullable
    -  _windchillm: type=utf8, nullable
    -  _wspdm: type=float64, nullable
  rows: 5
  col[0][datetime_utc]: ["19961101-17:00" "19961101-18:00" "19961101-19:00" "19961101-20:00" "19961101-21:00"]
  col[1][ _conds]: ["Smoke" "Smoke" "Smoke" "Smoke" "Smoke"]
  col[2][ _dewptm]: [12 13 13 13 13]
  col[3][ _fog]: [0 0 0 0 0]
  col[4][ _hail]: [0 0 0 0 0]
  col[5][ _heatindexm]: [(null) (null) (null) (null) (null)]
  col[6][ _hum]: [56 60 60 68 68]
  col[7][ _precipm]: [(null) (null) (null) (null) (null)]
  col[8][ _pressurem]: [1011 1010 -9999 -9999 1010]
  col[9][ _rain]: [0 0 0 0 0]
  col[10][ _snow]: [0 0 0 0 0]
  col[11][ _tempm]: [21 21 21 19 19]
  col[12][ _thunder]: [0 0 0 0 0]
  col[13][ _tornado]: [0 0 0 0 0]
  col[14][ _vism]: [(null) 0.8 (null) (null) (null)]
  col[15][ _wdird]: [0 0 0 0 0]
  col[16][ _wdire]: ["North" "North" "North" "North" "North"]
  col[17][ _wgustm]: [(null) (null) (null) (null) (null)]
  col[18][ _windchillm]: [(null) (null) (null) (null) (null)]
  col[19][ _wspdm]: [(null) 0 (null) (null) (null)]

record:
  schema:
  fields: 20
    - datetime_utc: type=utf8, nullable
    -  _conds: type=utf8, nullable
    -  _dewptm: type=int64, nullable
    -  _fog: type=int64, nullable
    -  _hail: type=int64, nullable
    -  _heatindexm: type=utf8, nullable
    -  _hum: type=int64, nullable
    -  _precipm: type=utf8, nullable
    -  _pressurem: type=int64, nullable
    -  _rain: type=int64, nullable
    -  _snow: type=int64, nullable
    -  _tempm: type=int64, nullable
    -  _thunder: type=int64, nullable
    -  _tornado: type=int64, nullable
    -  _vism: type=float64, nullable
    -  _wdird: type=int64, nullable
    -  _wdire: type=utf8, nullable
    -  _wgustm: type=utf8, nullable
    -  _windchillm: type=utf8, nullable
    -  _wspdm: type=float64, nullable
  rows: 2
  col[0][datetime_utc]: ["19961101-22:00" "19961101-23:00"]
  col[1][ _conds]: ["Smoke" "Smoke"]
  col[2][ _dewptm]: [13 12]
  col[3][ _fog]: [0 0]
  col[4][ _hail]: [0 0]
  col[5][ _heatindexm]: [(null) (null)]
  col[6][ _hum]: [68 64]
  col[7][ _precipm]: [(null) (null)]
  col[8][ _pressurem]: [1009 1009]
  col[9][ _rain]: [0 0]
  col[10][ _snow]: [0 0]
  col[11][ _tempm]: [19 19]
  col[12][ _thunder]: [0 0]
  col[13][ _tornado]: [0 0]
  col[14][ _vism]: [(null) (null)]
  col[15][ _wdird]: [0 0]
  col[16][ _wdire]: ["North" "North"]
  col[17][ _wgustm]: [(null) (null)]
  col[18][ _windchillm]: [(null) (null)]
  col[19][ _wspdm]: [(null) (null)]

这次12行数据都被成功读取出来了!

接下来,我们再来读取一下完整数据集testset.csv,我们通过输出读取的数据集行数来判断一下读取是否完全成功:

// read_csv_rows_count.go

func read(data io.ReadCloser) error {
    var total int64
    // read 10000 lines at a time to create record batches
    rdr := csv.NewInferringReader(data, csv.WithChunk(10000),
        // strings can be null, and these are the values
        // to consider as null
        csv.WithNullReader(true, "", "null", "[]"),
        // assume the first line is a header line which names the columns
        csv.WithHeader(true),
        csv.WithColumnTypes(map[string]arrow.DataType{
            " _vism": arrow.PrimitiveTypes.Float64,
        }),
    )

    for rdr.Next() {
        rec := rdr.Record()
        total += rec.NumRows()
    }

    fmt.Println("total columns =", total)
    return nil
}

我们开着错误输出的调试语句,看看上面的代码的输出结果:

======nextn: strconv.ParseInt: parsing "N/A": invalid syntax
total columns = 10000

我们看到上述程序仅读取了1w条记录,并输出了一个错误信息:CSV文件中包含“N/A”字样的数据,导致CSV Reader读取失败。经过数据对比核查,发现hum的数据存在大量“N/A”,另外pressurem的类型也有问题。那么如何解决这个问题呢?NewInferringReader提供了WithIncludeColumns功能选项可以供我们提供我们想要的列,这样我们可以给出一个列白名单,将hum列排除在外。修改后的read代码如下:

// read_csv_rows_count_with_col_filter.go

func read(data io.ReadCloser) error {
    var total int64
    // read 10000 lines at a time to create record batches
    rdr := csv.NewInferringReader(data, csv.WithChunk(10000),
        // strings can be null, and these are the values
        // to consider as null
        csv.WithNullReader(true, "", "null", "[]"),
        // assume the first line is a header line which names the columns
        csv.WithHeader(true),
        csv.WithColumnTypes(map[string]arrow.DataType{
            " _pressurem": arrow.PrimitiveTypes.Float64,
        }),
        csv.WithIncludeColumns([]string{
            "datetime_utc", // 19961101-11:00
            " _conds",      // Smoke、Haze
            " _fog",        // 0
            " _heatindexm",
            " _pressurem", //
            " _rain",      //
            " _snow",      //
            " _tempm",     //
            " _thunder",   //
            " _tornado",   //
        }),
    )   

    for rdr.Next() {
        rec := rdr.Record()
        total += rec.NumRows()
    }   

    fmt.Println("total columns =", total)
    return nil
}

运行修改后的代码:

$go run read_csv_rows_count_with_col_filter.go
total columns = 100990

我们顺利将CSV中的数据读到了内存中,并组织成了多个Record Batch。

2. Arrow compute API介绍

一旦内存中有了Arrow格式的数据后,我们就可以基于这份数据进行数据操作了,比如过滤、查询、计算、转换等等。那么是否需要开发人员自己根据对Arrow type的结构的理解来实现针对这些数据操作的算法呢?不用的

Arrow社区提供了compute API以及各种语言的高性能实现以供基于Arrow格式进行数据操作的开发人员直接复用。

Go Arrow实现也提供了compute包用于操作内存中的Arrow object。不过根据compute包的注释来看,目前Go compute包还属于实验性质,并非stable的API,将来可能有变:

// The overwhemling majority of things in this package require go1.18 as
// it utilizes generics. The files in this package and its sub-packages
// are all excluded from being built by go versions lower than 1.18 so
// that the larger Arrow module itself is still compatible with go1.17.
//
// Everything in this package should be considered Experimental for now.
package compute

另外我们从上面注释也可以看到,compute包用到了泛型,因此需要Go 1.18及以后版本才能使用。

为了更好地理解compute API,我们需要知道一些有关compute的概念,首先了解一下Datum。

2.1 Datum

compute API中的函数需要支持多种类型数据作为输入,可以是arrow的array type,也可以是一个标量值(scalar),为了统一输出表示,compute API建立了一个名为Datum的抽象。Datum可以理解为一个compute API函数可以接受的各种arrow类型的union类型,union中既可以是一个scalar(标量值),也可以是Array、Chunked Array,甚至是一整个Record Batch或一个Arrow Table。

不出预料,Go中采用接口来建立Datum这个抽象:

// Datum is a variant interface for wrapping the various Arrow data structures
// for now the various Datum types just hold a Value which is the type they
// are wrapping, but it might make sense in the future for those types
// to actually be aliases or embed their types instead. Not sure yet.
type Datum interface {
    fmt.Stringer
    Kind() DatumKind
    Len() int64
    Equals(Datum) bool
    Release()

    data() any
}

Datum支持的类型通过DatumKind的常量枚举值可以看出:

// DatumKind is an enum used for denoting which kind of type a datum is encapsulating
type DatumKind int

const (
    KindNone    DatumKind = iota // none
    KindScalar                   // scalar
    KindArray                    // array
    KindChunked                  // chunked_array
    KindRecord                   // record_batch
    KindTable                    // table
)

2.2 Function Type

compute包提供的是协助数据操作和分析的函数,这些函数可以被分为几类,我们由简单到复制的顺序逐一看一下:

2.2.1 标量(scalar)函数或逐元素(element-wise)函数

这类函数接受一个scalar参数或一个array类型的datum参数,函数会对输入参数中的逐个元素进行操作,比如求反、求绝对值等。如果传入的是scalar,则返回scalar,如果传入的是array类型,则返回array类型。传入和返回的array长度应相同。

下图(来自《In-Memory Analytics with Apache Arrow》一书)直观地解释了这类函数的操作特性:

我们用go代码实现一下上图中的两个示例,先来看unary element-wise操作的例子:

// unary_elementwise_function.go

func main() {
    data := []int32{5, 10, 0, 25, 2}
    bldr := array.NewInt32Builder(memory.DefaultAllocator)
    defer bldr.Release()
    bldr.AppendValues(data, nil)
    arr := bldr.NewArray()
    defer arr.Release()

    dat, err := compute.Negate(context.Background(), compute.ArithmeticOptions{}, compute.NewDatum(arr))
    if err != nil {
        fmt.Println(err)
        return
    }

    arr1, ok := dat.(*compute.ArrayDatum)
    if !ok {
        fmt.Println("type assert fail")
        return
    }
    fmt.Println(arr1.MakeArray()) // [-5 -10 0 -25 -2]
}

compute包实现了常见的一元和二元arithmetic函数:

下面是二元Add操作的示例:

// binary_elementwise_function.go

func main() {
    data1 := []int32{5, 10, 0, 25, 2}
    data2 := []int32{1, 5, 2, 10, 5}
    scalarData1 := int32(6)

    bldr := array.NewInt32Builder(memory.DefaultAllocator)
    defer bldr.Release()
    bldr.AppendValues(data1, nil)
    arr1 := bldr.NewArray()
    defer arr1.Release()

    bldr.AppendValues(data2, nil)
    arr2 := bldr.NewArray()
    defer arr2.Release()

    result1, err := compute.Add(context.Background(), compute.ArithmeticOptions{},
        compute.NewDatum(arr1),
        compute.NewDatum(arr2))
    if err != nil {
        fmt.Println(err)
        return
    }

    result2, err := compute.Add(context.Background(), compute.ArithmeticOptions{},
        compute.NewDatum(arr1),
        compute.NewDatum(scalarData1))
    if err != nil {
        fmt.Println(err)
        return
    }

    resultArr1, ok := result1.(*compute.ArrayDatum)
    if !ok {
        fmt.Println("type assert fail")
        return
    }
    fmt.Println(resultArr1.MakeArray()) // [6 15 2 35 7]

    resultArr2, ok := result2.(*compute.ArrayDatum)
    if !ok {
        fmt.Println("type assert fail")
        return
    }
    fmt.Println(resultArr2.MakeArray()) // [11 16 6 31 8]
}

在这个示例里,我们实现了array + array和array + scalar两个操作,两个加法操作的结果都是一个新array。

接下来我们来看

2.2.2 array-wise(逐array)函数

这一类的函数使用整个array进行操作,经常进行转换或输出与输入array不同长度的结果。下图(来自《In-Memory Analytics with Apache Arrow》一书)直观地解释了这类函数的操作特性:

Go compute包没有提供sort_unique函数,这里用Unique模拟一个unary array-wise操作,代码如下:

// unary_arraywise_function.go

func main() {
    data := []int32{5, 10, 0, 25, 2, 10, 2, 25}
    bldr := array.NewInt32Builder(memory.DefaultAllocator)
    defer bldr.Release()
    bldr.AppendValues(data, nil)
    arr := bldr.NewArray()
    defer arr.Release()

    dat, err := compute.Unique(context.Background(), compute.NewDatum(arr))
    if err != nil {
        fmt.Println(err)
        return
    }

    arr1, ok := dat.(*compute.ArrayDatum)
    if !ok {
        fmt.Println("type assert fail")
        return
    }
    fmt.Println(arr1.MakeArray()) // [5 10 0 25 2]
}

而上图中的二元array-wise Filter操作可以由下面代码实现:

// binary_arraywise_function.go

func main() {
    data := []int32{5, 10, 0, 25, 2}
    filterMask := []bool{true, false, true, false, true}

    bldr := array.NewInt32Builder(memory.DefaultAllocator)
    defer bldr.Release()
    bldr.AppendValues(data, nil)
    arr := bldr.NewArray()
    defer arr.Release()

    bldr1 := array.NewBooleanBuilder(memory.DefaultAllocator)
    defer bldr1.Release()
    bldr1.AppendValues(filterMask, nil)
    filterArr := bldr1.NewArray()
    defer filterArr.Release()

    dat, err := compute.Filter(context.Background(), compute.NewDatum(arr),
        compute.NewDatum(filterArr),
        compute.FilterOptions{})
    if err != nil {
        fmt.Println(err)
        return
    }

    arr1, ok := dat.(*compute.ArrayDatum)
    if !ok {
        fmt.Println("type assert fail")
        return
    }
    fmt.Println(arr1.MakeArray()) // [5 0 2]
}

注意:compute.Filter函数要求传入的value datum和filter datum的底层array长度要相同。

2.2.3 聚合(Aggregation)函数

Arrow compute支持两类聚合函数,一类是标量聚合(scalar aggregation),它的操作对象为一个array或一个标量,计算后输出一个标量值,常见的例子包括:count、min、max、mean、avg、sum等聚合计算;另外一类则是分组聚合(grouped aggregation),即先按某些“key”列进行分组后,再分别聚合,有些类似SQL中的group by操作。下图(来自《In-Memory Analytics with Apache Arrow》一书)直观地解释了这两类函数的操作特性:

不过遗憾的是Go尚未提供对这类聚合函数的直接支持

要想实现上述十分有用的聚合数据操作,在官方尚未提供支持之前,我们可以考虑自行扩展compute包。

注:相对完整的标量聚合和分组聚合的函数列表,可以参考C++版本的API ref

3. 小结

鉴于本篇篇幅以及Go对聚合函数的尚未支持,计划中对Delhi CSV文件的聚合分析只能留到后面系列文章了。

简单回顾一下本文内容。我们介绍了Go Arrow实现从CSV文件读取数据的方法以及一些技巧,然后我们介绍了Arrow除了其format之外的一个重点内容:compute API,这为基于arrow的array数据进行数据操作提供了开箱即用和高性能的API,大家要理解其中Datum的抽象概念,以及各类Function的操作对象和返回结果类型。

注:本文涉及的源代码在这里可以下载。

4. 参考资料

  • Make Data Files Easier to Work With Using Golang and Apache Arrow – https://voltrondata.com/resources/make-data-files-easier-to-work-with-golang-arrow
  • 《In-Memory Analytics with Apache Arrow》- https://book.douban.com/subject/35954154/
  • C++ compute API – https://arrow.apache.org/docs/cpp/compute.html
  • C++ Acero高级API – https://arrow.apache.org/docs/cpp/streaming_execution.html

“Gopher部落”知识星球旨在打造一个精品Go学习和进阶社群!高品质首发Go技术文章,“三天”首发阅读权,每年两期Go语言发展现状分析,每天提前1小时阅读到新鲜的Gopher日报,网课、技术专栏、图书内容前瞻,六小时内必答保证等满足你关于Go语言生态的所有需求!2023年,Gopher部落将进一步聚焦于如何编写雅、地道、可读、可测试的Go代码,关注代码质量并深入理解Go核心技术,并继续加强与星友的互动。欢迎大家加入!

img{512x368}
img{512x368}

img{512x368}
img{512x368}

著名云主机服务厂商DigitalOcean发布最新的主机计划,入门级Droplet配置升级为:1 core CPU、1G内存、25G高速SSD,价格5$/月。有使用DigitalOcean需求的朋友,可以打开这个链接地址:https://m.do.co/c/bff6eed92687 开启你的DO主机之路。

Gopher Daily(Gopher每日新闻)归档仓库 – https://github.com/bigwhite/gopherdaily

我的联系方式:

  • 微博(暂不可用):https://weibo.com/bigwhite20xx
  • 微博2:https://weibo.com/u/6484441286
  • 博客:tonybai.com
  • github: https://github.com/bigwhite

商务合作方式:撰稿、出书、培训、在线课程、合伙创业、咨询、广告合作。

如发现本站页面被黑,比如:挂载广告、挖矿等恶意代码,请朋友们及时联系我。十分感谢! Go语言第一课 Go语言精进之路1 Go语言精进之路2 商务合作请联系bigwhite.cn AT aliyun.com

欢迎使用邮件订阅我的博客

输入邮箱订阅本站,只要有新文章发布,就会第一时间发送邮件通知你哦!

这里是 Tony Bai的个人Blog,欢迎访问、订阅和留言! 订阅Feed请点击上面图片

如果您觉得这里的文章对您有帮助,请扫描上方二维码进行捐赠 ,加油后的Tony Bai将会为您呈现更多精彩的文章,谢谢!

如果您希望通过微信捐赠,请用微信客户端扫描下方赞赏码:

如果您希望通过比特币或以太币捐赠,可以扫描下方二维码:

比特币:

以太币:

如果您喜欢通过微信浏览本站内容,可以扫描下方二维码,订阅本站官方微信订阅号“iamtonybai”;点击二维码,可直达本人官方微博主页^_^:
本站Powered by Digital Ocean VPS。
选择Digital Ocean VPS主机,即可获得10美元现金充值,可 免费使用两个月哟! 著名主机提供商Linode 10$优惠码:linode10,在 这里注册即可免费获 得。阿里云推荐码: 1WFZ0V立享9折!


View Tony Bai's profile on LinkedIn
DigitalOcean Referral Badge

文章

评论

  • 正在加载...

分类

标签

归档



View My Stats