Deephaven Python客户端

Deephaven Python客户端是由Deephaven数据实验室创建的Python包。它是一个客户端API，允许Python应用程序远程访问Deephaven数据服务器。

源目录

从deephaven-core存储库根目录

(从https://github.com/deephaven/deephaven-core克隆)

切换到deephaven-core存储库内的py/client目录

$ cd py/client

开发环境设置

$ pip3 install -r requirements-dev.txt

构建

$ DEEPHAVEN_VERSION=$(../../gradlew :printVersion -q) python3 setup.py bdist_wheel

运行测试

$ python3 -m unittest discover tests

运行示例

$ python3 -m examples.demo_table_ops
$ python3 -m examples.demo_query
$ python3 -m examples.demo_run_script
$ python3 -m examples.demo_merge_tables
$ python3 -m examples.demo_asof_join

安装

注意实际的.whl文件名可能取决于系统详情。

$ pip3 install dist/pydeephaven-<x.y.z>-py3-none-any.whl

快速入门

    >>> from pydeephaven import Session
    >>> session = Session() # assuming Deephaven Community Edition is running locally with the default configuration
    >>> table1 = session.time_table(period=1000000000).update(formulas=["Col1 = i % 2"])
    >>> df = table1.to_arrow().to_pandas()
    >>> print(df)
                        Timestamp  Col1
    0     1629681525690000000     0
    1     1629681525700000000     1
    2     1629681525710000000     0
    3     1629681525720000000     1
    4     1629681525730000000     0
    ...                   ...   ...
    1498  1629681540670000000     0
    1499  1629681540680000000     1
    1500  1629681540690000000     0
    1501  1629681540700000000     1
    1502  1629681540710000000     0

    >>> session.close()

初始化

Session类是您连接到Deephaven的方式。这是您的Python代码与Deephaven服务器交互的方式。

from pydeephaven import Session

session = Session()

定时表

Session类有许多创建表的方法。此示例创建一个定时时间表并将其绑定到Deephaven。

from pydeephaven import Session

session = Session()

table = session.time_table(period=1000000000).update(formulas=["Col1 = i % 2"])
session.bind_table(name="my_table", table=table)

这是Python客户端与Deephaven交互的一般流程。您创建一个表（新或现有），在它上面执行一些操作，然后将其绑定到Deephaven。绑定表在Deephaven服务器上给它一个命名引用，以便可以从Web API或其他会话中使用它。

在表上执行查询

table.update() 可以用来对表执行更新。这通过查询字符串来更新表。

from pydeephaven import Session

session = Session()

# Create a table with no columns and 3 rows
table = session.empty_table(3)
# Create derived table having a new column MyColumn populated with the row index "i"
table = table.update(["MyColumn = i"])
# Update the Deephaven Web Console with this new table
session.bind_table(name="my_table", table=table)

对表进行排序

table.sort() 可以用来对表进行排序。以下示例展示了如何通过表的一列进行排序。

from pydeephaven import Session

session = Session()

table = session.empty_table(5)
table = table.update(["SortColumn = 4-i"])

table = table.sort(["SortColumn"])
session.bind_table(name="my_table", table=table)

过滤表

table.where() 可以用来过滤表。以下示例展示了如何使用过滤字符串来过滤表。

from pydeephaven import Session

session = Session()

table = session.empty_table(5)
table = table.update(["Values = i"])

table = table.where(["Values % 2 == 1"])
session.bind_table(name="my_table", table=table)

查询对象

查询对象是一种创建和管理一系列Deephaven查询操作的方式，作为一个单独的单元。查询对象有可能比相应的单个查询表现更好，因为查询对象可以一次性发送到服务器，而不是多个请求，并且当系统能够一次性看到整个查询序列时，可以执行某些优化。它们在精神上类似于SQL中的预处理语句。

使用查询对象的一般流程是：使用表构建查询，在查询对象上调用表操作（排序、过滤、更新等），然后将您的表分配给query.exec()的返回值。

任何可以在表上执行的操作也可以在查询对象上执行。以下示例展示了两个计算相同结果的操作，第一个使用表更新，第二个使用查询对象。

from pydeephaven import Session

session = Session()

table = session.empty_table(10)

# executed immediately
table1= table.update(["MyColumn = i"]).sort(["MyColumn"]).where(["MyColumn > 5"]);

# create Query Object (execution is deferred until the "exec" statement)
query_obj = session.query(table).update(["MyColumn = i"]).sort(["MyColumn"]).where(["MyColumn > 5"]);
# Transmit the QueryObject to the server and execute it
table2 = query_obj.exec();

session.bind_table(name="my_table1", table=table1)
session.bind_table(name="my_table2", table=table2)

连接两个表

table.join() 是许多可以将两个表连接起来的操作之一，如下所示。

from pydeephaven import Session

session = Session()

table1 = session.empty_table(5)
table1 = table1.update(["Values1 = i", "Group = i"])
table2 = session.empty_table(5)
table2 = table2.update(["Values2 = i + 10", "Group = i"])

table = table1.join(table2, on=["Group"])
session.bind_table(name="my_table", table=table)

在表上执行聚合操作

在Python客户端中可以对表应用聚合。以下示例创建了一个聚合，它计算表中的Count列的平均值，并按Group列进行聚合。

from pydeephaven import Session, agg

session = Session()

table = session.empty_table(10)
table = table.update(["Count = i", "Group = i % 2"])

my_agg = agg.avg(["Count"])

table = table.agg_by(aggs=[my_agg], by=["Group"])
session.bind_table(name="my_table", table=table)

将PyArrow表转换为Deephaven表

Deephaven原生支持PyArrow表。以下示例展示了如何在PyArrow表和Deephaven表之间进行转换。

import pyarrow as pa
from pydeephaven import Session

session = Session()

arr = pa.array([4,5,6], type=pa.int32())
pa_table = pa.Table.from_arrays([arr], names=["Integers"])

table = session.import_table(pa_table)
session.bind_table(name="my_table", table=table)

#Convert the Deephaven table back to a pyarrow table
pa_table = table.to_arrow()

在服务器端执行脚本

session.run_script() 可以用来在Deephaven服务器上执行代码。当操作无法在客户端执行时，例如创建动态表写入器，这非常有用。以下示例展示了如何在服务器端执行脚本并检索由脚本生成的表。

from pydeephaven import Session

session = Session()

script = """
from deephaven import empty_table

table = empty_table(8).update(["Index = i"])
"""

session.run_script(script)

table = session.open_table("table")
print(table.to_arrow())

错误处理

当客户端包遇到错误时，会抛出DHError。以下示例展示了如何捕获DHError。

from pydeephaven import Session, DHError

try:
    session = Session(host="invalid_host")
except DHError as e:
    print("Deephaven error when connecting to session")
    print(e)
except Exception as e:
    print("Unknown error")
    print(e)

相关文档

• https://deephaven.io/
• https://arrow.apache.ac.cn/docs/python/index.html

API参考

[从这里开始] https://deephaven.io/core/client-api/python/