Skip to content

简介

logo

什么是ArcticDB?

ArcticDB 是一个无服务器的 DataFrame 数据库引擎,旨在为 Python 数据科学生态系统服务。

ArcticDB使您能够以规模存储、检索和处理数据框,得益于商品对象存储(与S3兼容的存储和Azure Blob存储)。

ArcticDB需要零附加基础设施,只需运行Python环境和访问对象存储即可,并且可以在几秒钟内安装。

ArcticDB 是:

  • 快速
    • 单个消费者每秒处理高达1亿行数据
    • 所有消费者每秒处理10亿行数据
    • 安装简单快捷: pip install arcticdb
  • 灵活
    • 不需要数据模式
    • 支持流数据摄取
    • 双时态 - 存储所有存储数据的先前版本
    • 本地和云端都容易设置
    • 从开发/研究到生产环境可扩展
  • 熟悉
    • ArcticDB 是世界上最简单的可共享数据库
    • 对于拥有 Python 和 Pandas 经验的人来说,容易学习
    • 只有你和你的数据 - 认知负担非常低。

开始

本节将涵盖安装、设置和基本使用。有关基础和高级功能的更多详细信息,请参阅教程部分。

安装

ArcticDB 支持 Python 3.6 - 3.11。Python 3.7 是 Windows 的最早版本。

要安装,只需运行:

pip install arcticdb

设置

ArcticDB 是一个为对象存储设计的存储引擎,同时使用 LMDB 也支持本地磁盘存储。

存储

ArcticDB 支持任何兼容 S3 API 的存储,包括 AWS 和 Azure,以及像 VAST Universal StoragePure Storage 的存储设备。

ArcticDB 还支持 LMDB 作为本地/文件基础存储 - 要使用 LMDB,请将 LMDB 路径作为 URI 传入: adb.Arctic('lmdb://path/to/desired/database')

要开始,我们可以导入ArcticDB并实例化它:

import arcticdb as adb
# this will set up the storage using the local file system
uri = "lmdb://tmp/arcticdb_intro"
ac = adb.Arctic(uri)

有关如何正确格式化其他存储的 uri 字符串的更多信息,请查看文档字符串 (help(Arctic)) 或阅读 存储访问 部分(点击链接或继续阅读本节下面的内容)。

库设置

ArcticDB旨在存储许多(可能是数百万个)表。单个表(数据框)称为符号,并存储在称为的集合中。单个可以存储多个符号。

在使用之前必须先初始化:

ac.create_library('intro')  # 静态模式 - 请参见下面的说明
ac.list_libraries()
output
['intro']

然后必须在代码中实例化库,以便读取/写入数据:

library = ac['intro']

有时候,使用这种形式结合库创建和实例化更方便,这将自动创建库(如果需要的话),以节省您检查它是否已经存在的时间:

library = ac.get_library('intro', create_if_missing=True)

极地数据库静态与动态模式

ArcticDB 不需要数据模式,与许多其他数据库不同。您可以编写任何 DataFrame 并随后读取。如果数据的形状发生变化然后再次写入,它将全部正常工作。简单明了。

需要使用模式的唯一例外是在修改现有符号的函数的情况下: updateappend。在修改符号时,新数据必须与现有数据具有相同的模式。这里的模式意味着索引类型以及 DataFrame 中每一列的名称、顺序和类型。换句话说,当您要追加新行时,它们必须看起来像现有行。这是默认选项,称为 static schema

然而,如果您需要通过 updateappend 添加、删除或更改列的类型,那么您可以这样做。您只需创建库并设置 dynamic_schema 选项。请参见 (create_library) 方法的 library_options 参数。

所以你可以同时拥有两个优点 - 你可以选择对你的数据施加静态模式,这样通过修改操作就不能更改它,或允许它保持灵活。

在库创建时必须选择使用静态或动态模式。

在本节中,我们使用 static schema,以便清楚明了。

读取和写入数据

现在我们已经设置了一个库,我们可以开始读取和写入数据。ArcticDB 有一组用于 DataFrame 存储的简单函数。

让我们将一个数据框写入存储。

首先创建数据:

# 50 列,25 行,随机数据,按日期时间索引。
import pandas as pd
import numpy as np
from datetime import datetime
cols = ['COL_%d' % i for i in range(50)]
df = pd.DataFrame(np.random.randint(0, 50, size=(25, 50)), columns=cols)
df.index = pd.date_range(datetime(2000, 1, 1, 5), periods=25, freq="h")
df.head(5)
输出(数据的前5行)
                     COL_0  COL_1  COL_2  COL_3  COL_4  COL_5  COL_6  COL_7  ...
2000-01-01 05:00:00     18     48     10     16     38     34     25     44  ...
2000-01-01 06:00:00     48     10     24     45     22     36     30     19  ...
2000-01-01 07:00:00     25     16     36     29     25      9     48      2  ...
2000-01-01 08:00:00     38     21      2      1     37      6     31     31  ...
2000-01-01 09:00:00     45     17     39     47     47     11     33     31  ...

然后写入库:

library.write('test_frame', df)
输出(关于写入内容的信息)
VersionedItem(symbol=test_frame,library=intro,data=n/a,version=0,metadata=None,host=)

'test_frame' DataFrame 将在本指南的其余部分中使用。

北极数据库索引

在编写Pandas DataFrames时,ArcticDB支持以下索引类型:

  • pandas.Index 包含 int64(或对应的专用类型 Int64IndexUInt64Index
  • RangeIndex
  • DatetimeIndex
  • MultiIndex 由上述支持的类型组成

head()/tail()中的“行”概念指的是行号('iloc'),而不是pandas.Index中的值('loc')。

从存储中读取数据:

from_storage_df = library.read('test_frame').data
from_storage_df.head(5)
输出(前5行数据,但从数据库读取)
                     COL_0  COL_1  COL_2  COL_3  COL_4  COL_5  COL_6  COL_7  ...
2000-01-01 05:00:00     18     48     10     16     38     34     25     44  ...
2000-01-01 06:00:00     48     10     24     45     22     36     30     19  ...
2000-01-01 07:00:00     25     16     36     29     25      9     48      2  ...
2000-01-01 08:00:00     38     21      2      1     37      6     31     31  ...
2000-01-01 09:00:00     45     17     39     47     47     11     33     31  ...

读取的数据与原始数据匹配,当然。

切片和过滤

ArcticDB 使您能够按行和按列进行切片。

北极数据库索引

ArcticDB将为有序的数值和时间序列(例如,DatetimeIndex)Pandas索引构建一个完整的索引。这将使得 在索引条目之间进行优化切片成为可能。如果索引是无序的或不是数字类型,您的数据仍然可以存储,但行切片的速度会 变慢。

行切片

library.read('test_frame', date_range=(df.index[5], df.index[8])).data
输出(请求的数据范围内的行)
                     COL_0  COL_1  COL_2  COL_3  COL_4  COL_5  COL_6  COL_7  ...
2000-01-01 10:00:00     23     39      0     45     15     28     10     17  ...
2000-01-01 11:00:00     36     28     22     43     23      6     10      1  ...
2000-01-01 12:00:00     18     42      1     15     19     36     41     36  ...
2000-01-01 13:00:00     28     32     47     37     17     44     29     24  ...

列切片

_range = (df.index[5], df.index[8])
_columns = ['COL_30', 'COL_31']
library.read('test_frame', date_range=_range, columns=_columns).data
输出(在指定日期范围和请求的列中的行)
                     COL_30  COL_31
2000-01-01 10:00:00      31       2
2000-01-01 11:00:00       3      34
2000-01-01 12:00:00      24      43
2000-01-01 13:00:00      18       8

过滤和分析

ArcticDB支持许多常见的DataFrame分析操作,包括过滤、投影、分组、聚合和重采样。访问这些操作的最直观方式是通过LazyDataFrame API,对于熟悉Pandas或Polars的用户来说应该感觉很熟悉。

遗留的 QueryBuilder 类也可以直接创建并传入 read 调用,效果相同。

北极数据库分析哲学

在大多数情况下,这比等效的Pandas操作更节省内存和更高效,因为处理是在C++存储引擎中进行的,并且在多个执行线程上并行化。

import arcticdb as adb
_range = (df.index[5], df.index[8])
_cols = ['COL_30', 'COL_31']
# 使用惰性计算
lazy_df = library.read('test_frame', date_range=_range, columns=_cols, lazy=True)
lazy_df = lazy_df[(lazy_df["COL_30"] > 10) & (lazy_df["COL_31"] < 40)]
df = lazy_df.collect().data
# 使用遗留的 QueryBuilder 类给出相同的结果
q = adb.QueryBuilder()
q = q[(q["COL_30"] > 10) & (q["COL_31"] < 40)]
library.read('test_frame', date_range=_range, columns=_cols, query_builder=q).data
输出(根据日期范围、列和基于数据值的查询过滤的数据)
                     COL_30  COL_31
2000-01-01 10:00:00      31       2
2000-01-01 13:00:00      18       8

修改,版本控制(即时间旅行)

ArcticDB 完全支持通过两个原语修改存储的数据:updateappend

这些操作是原子的,但不会锁定符号。有关更多信息,请参阅事务部分。

追加

让我们将数据附加到时间序列的末尾。

首先,我们将查看数据的最后几条记录(在它被修改之前)

library.tail('test_frame', 4).data
输出
                     COL_0  COL_1  COL_2  COL_3  COL_4  COL_5  COL_6  COL_7  ...
2000-01-02 02:00:00     46     12     38     47      4     31      1     42  ...
2000-01-02 03:00:00     46     20      5     42      8     35     12      2  ...
2000-01-02 04:00:00     17     48     36     43      6     46      5      8  ...
2000-01-02 05:00:00     20     19     24     44     29     32      2     19  ...
然后创建 3 行新数据进行追加。要使追加操作有效,新数据的第一个 datetime 必须在现有数据之后开始。

random_data = np.random.randint(0, 50, size=(3, 50))
df_append = pd.DataFrame(random_data, columns=['COL_%d' % i for i in range(50)])
df_append.index = pd.date_range(datetime(2000, 1, 2, 7), periods=3, freq="h")
df_append
输出
                     COL_0  COL_1  COL_2  COL_3  COL_4  COL_5  COL_6  COL_7  ...
2000-01-02 07:00:00      9     15      4     48     48     35     34     49  ...
2000-01-02 08:00:00     35      4     12     30     30     12     38     25  ...
2000-01-02 09:00:00     25     17      3      1      1     15     33     49  ...

现在 append 这个数据框到之前写入的内容

library.append('test_frame', df_append)
输出
VersionedItem(symbol=test_frame,library=intro,data=n/a,version=1,metadata=None,host=)
然后查看最后5行以了解发生了什么

library.tail('test_frame', 5).data
输出
                     COL_0  COL_1  COL_2  COL_3  COL_4  COL_5  COL_6  COL_7  ...
2000-01-02 04:00:00     17     48     36     43      6     46      5      8  ...
2000-01-02 05:00:00     20     19     24     44     29     32      2     19  ...
2000-01-02 07:00:00      9     15      4     48     48     35     34     49  ...
2000-01-02 08:00:00     35      4     12     30     30     12     38     25  ...
2000-01-02 09:00:00     25     17      3      1      1     15     33     49  ...

最后的5行由之前写的最后两行和我们刚刚添加的3行组成。

Append对于将新数据添加到大型时间序列的末尾非常有用。

更新

更新原语使您能够覆盖一块连续的数据。这会导致修改某些行并删除其他行,正如我们在下面的示例中看到的。

在这里我们创建一个新的DataFrame用于更新,仅包含相隔2小时的2行

random_data = np.random.randint(0, 50, size=(2, 50))
df = pd.DataFrame(random_data, columns=['COL_%d' % i for i in range(50)])
df.index = pd.date_range(datetime(2000, 1, 1, 5), periods=2, freq="2h")
df
输出(仅选择行 0 和 2,由 iloc[] 选定)
                     COL_0  COL_1  COL_2  COL_3  COL_4  COL_5  COL_6  COL_7  ...
2000-01-01 05:00:00     47     49     15      6     22     48     45     22  ...
2000-01-01 07:00:00     46     10      2     49     24     49      8      0  ...
现在更新符号
library.update('test_frame', df)
输出(关于更新的信息)
VersionedItem(symbol=test_frame,library=intro,data=n/a,version=2,metadata=None,host=)

现在让我们来看一下符号中的前4行:

library.head('test_frame', 4).data  # head/tail 与等效的 Pandas 操作类似
output
                     COL_0  COL_1  COL_2  COL_3  COL_4  COL_5  COL_6  COL_7  ...
2000-01-01 05:00:00     47     49     15      6     22     48     45     22  ...
2000-01-01 07:00:00     46     10      2     49     24     49      8      0  ... 
2000-01-01 08:00:00     38     21      2      1     37      6     31     31  ... 
2000-01-01 09:00:00     45     17     39     47     47     11     33     31  ...

让我们来解析一下我们是如何得到这个结果的。更新内容是

  • 在索引匹配的情况下,用新数据替换符号中的数据(在这种情况下是 05:00 和 07:00 行)
  • 删除新数据的日期范围内不在新数据索引中的任何行(在这种情况下是06:00行)
  • 其余数据保持不变(在这种情况下为09:00及之后)

从逻辑上讲,这相当于用新数据替换旧数据的完整日期范围,这正是您对更新的期望。

版本管理

你可能注意到 read 调用并不会直接返回数据 - 而是返回一个 VersionedItem 结构。你可能还注意到修改操作 (writeappendupdate) 会递增版本号。ArcticDB 对所有修改进行版本控制,这意味着你可以检索早期版本的数据 - 它是一个双时间数据库:

library.tail('test_frame', 7, as_of=0).data
输出
                     COL_0  COL_1  COL_2  COL_3  COL_4  COL_5  COL_6  COL_7  ...
2000-01-01 23:00:00     16     46      3     45     43     14     10     27  ...
2000-01-02 00:00:00     37     37     20      3     49     38     23     46  ...
2000-01-02 01:00:00     42     47     40     27     49     41     11     26  ...
2000-01-02 02:00:00     46     12     38     47      4     31      1     42  ...
2000-01-02 03:00:00     46     20      5     42      8     35     12      2  ...
2000-01-02 04:00:00     17     48     36     43      6     46      5      8  ...
2000-01-02 05:00:00     20     19     24     44     29     32      2     19  ...

请注意时间戳 - 我们在append操作之前读取了数据。请注意,您还可以将datetime传递给任何as_of参数,这将导致读取早于传递的datetime的最后版本。

版本控制、修剪先前版本和快照

默认情况下,writeappendupdate 操作不会删除之前的版本。请注意,这将占用更多空间。

此行为可以通过 prune_previous_versions 关键字参数进行控制。将节省空间,但之前的版本将不可用。

可以通过使用快照来实现妥协,这允许将库的状态保存并在稍后读取。这允许某些版本受到删除的保护,当快照被删除时,它们将被删除。有关详细信息,请参见snapshot documentation

存储访问

S3 配置

配置S3访问有两种方法。如果您恰好知道访问密钥和秘密密钥,可以简单地按如下方式连接:

import arcticdb as adb
ac = adb.Arctic('s3://ENDPOINT:BUCKET?region=blah&access=ABCD&secret=DCBA')

否则,您可以将身份验证委托给 AWS SDK(遵循标准 AWS 配置选项):

ac = adb.Arctic('s3://ENDPOINT:BUCKET?aws_auth=true')

与上面相同,但使用HTTPS:

ac = adb.Arctic('s3s://ENDPOINT:BUCKET?aws_auth=true')

S3

如果您的 S3 端点使用 HTTPS,请使用 s3s

连接到定义的存储端点

使用预定义的访问和存储密钥连接到本地存储(不是 AWS - s3.local 的 HTTP 端点):

ac = adb.Arctic('s3://s3.local:arcticdb-test-bucket?access=EFGH&secret=HGFE')
连接到AWS

使用预定义区域连接到AWS:

ac = adb.Arctic('s3s://s3.eu-west-2.amazonaws.com:arcticdb-test-bucket?aws_auth=true')

请注意,没有给出显式的凭证参数。当 aws_auth 被传递时,身份验证委托给 AWS SDK,它负责在 .config 文件或环境变量中查找适当的凭证。您可以通过设置 AWS_PROFILE 环境变量来手动配置使用的配置文件,如在 AWS 文档 中所述。

在存储桶内使用特定路径

您可能想要将ArcticDB库的访问限制为桶内的特定路径。要做到这一点,您可以使用path_prefix参数:

ac = adb.Arctic('s3s://s3.eu-west-2.amazonaws.com:arcticdb-test-bucket?path_prefix=test&aws_auth=true')

Azure

ArcticDB使用Azure连接字符串来定义连接:

import arcticdb as adb
ac = adb.Arctic('azure://AccountName=ABCD;AccountKey=EFGH;BlobEndpoint=ENDPOINT;Container=CONTAINER')

例如:

import arcticdb as adb
ac = adb.Arctic("azure://CA_cert_path=/etc/ssl/certs/ca-certificates.crt;BlobEndpoint=https://arctic.blob.core.windows.net;Container=acblob;SharedAccessSignature=sp=awd&st=2001-01-01T00:00:00Z&se=2002-01-01T00:00:00Z&spr=https&rf=g&sig=awd%3D")

欲了解更多信息,请参见北极类参考

LMDB

LMDB支持配置其映射大小。请查看其文档

在Windows上您可能需要调整它,而在Linux上默认值要大得多,应当足够。这是因为Windows会急切地为映射文件分配物理空间,而在Linux上,映射大小是将要使用的物理空间的上限。

您可以在连接字符串中设置地图大小:

import arcticdb as adb
ac = adb.Arctic('lmdb://path/to/desired/database?map_size=2GB')

在Windows上的默认值是2GiB。与 lmdb errror code -30792 相关的错误表明映射正在变满,您应该增加其大小。如果您正在进行大规模写入,这种情况将会发生。

在每个Python进程中,您应该确保在给定的LMDB数据库上只打开一个Arctic实例。

LMDB不支持远程文件系统。

内存配置

提供的内存后端主要用于测试和实验。当不希望创建带有LMDB的文件时,它可能会很有用。

没有配置参数,内存完全由Arctic实例拥有。

例如:

import arcticdb as adb
ac = adb.Arctic('mem://')

对于对本地后端的并发访问,我们推荐使用连接到 tmpfs 的 LMDB,请参见 LMDB and In-Memory Tutorial

交易

  • 事务可以非常有用,但通常代价高昂且速度较慢
  • 如果我们拆分ACID:原子性、一致性和持久性是有用的,隔离性则少一些
  • 大多数分析工作流程可以构建为完全不需要事务运行
  • 那么,为什么要支付交易的成本,当它们通常是不需要的呢?
  • ArcticDB 没有事务,因为它是为高吞吐量的分析工作负载而设计的