k2magic 0.2.4

K2data内部的数据分析工具包

K2Magic

K2Magic是K2Assets提供的数据分析开发包（以下简称SDK），用于简化Python里访问各类数据库的操作。

一、安装

1.1 安装SDK

安装SDK最新版本：

pip install -U k2magic

1.2 安装数据库驱动

SDK兼容多种数据库的方言，用户只需修改初始化SDK时的连接字符串参数即可切换到另一个数据库。需要确保已安装此数据库的驱动包，例如对MySQL数据库需要pip install pymysql。

常见的数据库的驱动包和连接字符串如下:

数据库	驱动包	连接字符串	备注
PostgreSQL	psycopg2	postgresql+psycopg2://user:password@hostname:port/db_name
MySQL	pymysql	mysql+pymysql://user:password@hostname:port/db_name	未测试
Oracle	cx_oracle	oracle+cx_oracle://user:password@hostname:port/db_name	配置说明
SQL Server	pymssql	mssql+pymssql://user:password@hostname:port/db_name
K2Assets	无	k2assets://hostname:port/k2repo

表中未包含的数据库请参考：https://docs.sqlalchemy.org/en/20/dialects/

二、使用SDK

2.1 访问数据库

在python代码里使用SDK可以直接连接到指定数据库，并读写其中数据表的数据。假设数据库里有数据表table1的结构如下::

CREATE TABLE table1 (
    k_device VARCHAR(255) PRIMARY KEY,
    col1 FLOAT,
    col2 FLOAT,
    col3 FLOAT
);

对上述数据表操作的示例代码：

import pandas as pd
from k2magic.dataframe_db import DataFrameDB

db = DataFrameDB('postgresql+psycopg2://...')  # 此为连接字符串
df = pd.DataFrame({'k_device': ['a', 'b', 'c'], 'col1': [1, 2, 3], 'col2': [4, 5, 6]})

db.insert('table1', df)
db.update('table1', df, index_keys=['k_device'])
db.upsert('table1', df, index_keys=['k_device'])
df = db.select('table1', condition='col1 > 1')
df = db.select('table1', limit=3, order_by=['k_device DESC'])
df = db.sql_select('select * from table1')
db.delete('table1')

db.create_table(df, 'table2', primary_keys=['k_device'])
db.drop_table('table2')

2.2 访问K2Assets

除了能够直接访问各类数据库外，SDK还支持访问K2Assets平台的数据，即可以将K2Assets当作数据库，其中的Repo就是数据库里的表。

使用方法与其他数据库是相同的，只要更换连接字符串即可，下面是一个示例：

import pandas as pd
from k2magic.dataframe_db import DataFrameDB

# 连接到dev环境的K2Assets，每个repo视为一张表
db = DataFrameDB('k2assets://192.168.132.167:8765/k2repo')

# 查询数据（默认返回最新数据）
df = db.select('repo_XiLinHaoTe_1sec', columns=['k_device', 'k_ts', 'NacWdSpdFltS', 'CnvW'])

# 查询数据（带时间和设备过滤条件）
df = db.select('repo_XiLinHaoTe_1sec', columns=['k_device', 'k_ts', 'NacWdSpdFltS', 'CnvW'],
              condition='k_ts between 1656691200000 AND 1656777600000 and k_device = \'XiLinHaoTe002\'')

注意：

• 目前SDK只支持读取Repo数据，未来有可能支持写数据；
• K2Assets环境需要部署k2a-calcite-service服务，目前此服务还没有包含在K2Assets产品里，只在dev环境部署了用于测试。

2.3 异常处理

SDK进行数据库操作时，如果中途遇到异常情况会抛出DataFrameDBException，并自动回滚数据库到操作前的状态。用户可以使用try..except捕获此异常进行必要的处理：

try:
    db.insert('table1', df)
except DataFrameDBException as e:
    print('数据插入(部分)失败，操作已回滚')

三、常见问题

3.1 性能问题

通过SDK读取或写入大量数据时，性能会成为关键因素。一方面DataFrame对象过大可能导致内存溢出，另一方面若底层数据库的读写效率低可能导致操作超时。

解决这个问题的总体思路是将大的DataFrame分为小批处理，并且尽量对使用者保持透明，隐藏实现细节，降低对使用者的要求。

3.2 访问带schema前缀的表

有些数据库内部允许将数据表按schema进行划分，如果要访问非默认schema内部的表，初始化DataFrameDB对象时需要指定schema名称。例如：

db = DataFrameDB('mssql+pymssql://...', schema='schema1')
df = db.select('table1')   # 若表名不加schema前缀，会自动使用初始化时指定的schema作为前缀
df = db.select('schema1.table1')  # 使用完整表名也可以访问
df = db.select('schema2.table2')  # 访问同一数据库内其他schema的表也是允许的