使用 DolphinDB Class 来开发流计算状态算子

类的定义格式如下：

class 类名 {
  属性1 :: 类型1
  属性2 :: 类型2
  // ...
  // 和类名同名的方法成为构造函数，有且只有一个
  def 类名(arg1, arg2 /*,...*/) {
    属性1 = arg1
    属性2 = arg2
    // ...
  }
  // 需要注意，成员变量和成员函数不能同名。
  def 方法(arg1, arg2 /*, ...*/) {
    // ...
  }
}

例如，我们要定义一个 Person 类，包含两个成员变量：name 和 age。其中，name 是字符串，age 是整数。该类还包含了一个构造函数和 name 成员变量的 getter/setter。

定义如下：

class Person {
  // 变量声明在方法声明之前
  name :: STRING
  age :: INT
  // 定义构造函数
  def Person(name_, age_) { // 参数名不能和属性名相同，否则会覆盖属性名
    name = name_
    age = age_
  }
  def setName(newName) {
    name = newName
  }
  def getName() {
    return name
  }
}

可以通过 object.method() 的形式调用对象的成员函数；通过 object.member 的形式访问对象的属性。需要注意，和 Python 等脚本语言不同，无法直接通过对 object.member 赋值修改成员变量。如果需要为成员变量赋值，需要创建并使用相应的 setter 方法进行赋值。

p = Person("Zhang San", 30)
print(p.getName())
// 调用对象的方法
p.setName("Li Si")
print(p.getName())
// 引用对象的属性
print(p.name)
p.name = "Wang Wu" // 报错：禁止对对象的属性直接赋值

定义成员变量的格式为：成员变量名 :: 类型标注。

其中类型标注可以是：

• 标量：包括所有的基本类型，如 INT, DOUBLE, STRING，以及时间类型。
• 向量：如 DOUBLE VECTOR, STRING VECTOR 或者 Array Vector：INT[] VECTOR。
• 其他形式/类型不限：如果是其他类型（如字典、函数等），或者不希望限定变量类型，可以把类型标注写成 ANY。

例如：

a :: INT
b :: DOUBLE VECTOR
handler :: ANY

方法中使用到的变量的解析顺序：

1. 方法参数
2. 对象属性
3. 共享变量

share table(1:0, `sym`val, [SYMBOL, INT]) as tbl
go
class Test2 {
  a :: INT
  b :: DOUBLE
  def Test2() {
    a = 1
    b = 2.0
  }
  def method(b) {
    print(b) // 解析为函数参数 b；如果需要访问成员变量 b，需要使用 self，见下一小节
    print(a) // 解析为对象属性 a
    print(tbl) // 解析为共享变量 tbl
    print(qwert) // 变量不存在，报错
  }
}

通过 “self” 变量在类的方法中获取对象本身，其类似于 Python 中的 self，或者 Java、C++ 中的 this。

def doSomething(a) {
  // ...
}

class A{
	a :: INT
	b :: INT
	def A() {
		a = 1
		b = 2
	}
	def createCallback() {
		return doSomething{self}
	}
	def nameShadow(b) {
		print(b)
		print(self.b)
	}
}
a = A()
handler = a.createCallback()
a.nameShadow(3)

在状态引擎内部，cumsum 算子实现了累计求和的功能。这个功能在状态引擎内的实现非常简单，本节不展开说明，仅说明如何利用 OOP 重新实现该算子。

首先，定义一个类 MyCumSum，并将算子的状态定义为类的成员变量。

其次，在类中实现 append 方法，用于实现累计求和的功能。append 的参数是逐行输入的数据，返回的结果将作为计算结果输出。

最后，定义一个状态引擎，并指定 MyCumSum.append() 为引擎的算子。向引擎中输入数据，并查看计算结果。

代码如下：

class MyCumSum {
  sum :: DOUBLE
  def MyCumSum() {
    sum = 0.0
  }
  def append(value) {
    sum = sum + value
    return sum
  }
}

inputTable = table(1:0, `sym`val, [SYMBOL, DOUBLE])
result = table(1000:0, `sym`res, [SYMBOL, DOUBLE])

rse = createReactiveStateEngine(
          name="reactiveDemo",
          metrics = [<MyCumSum().append(val)>],
          dummyTable=inputTable,
          outputTable=result,
          keyColumn="sym")

data = table(take(`A, 100) as sym, rand(100.0, 100) as val)
rse.append!(data)

select * from data
select * from result

进行一次运行，随机生成出来的输入数据和对应输出为：

输入数据

输出数据

可以看到，我们编写的 OOP 算子实现了分组的累加求和，与状态引擎内置的 cumsum 算子的功能一致。

在未提供 OOP 时，状态引擎计算线性递归时需要通过内置函数 stateIterate 实现。使用stateIterate 需要指定用于迭代计算的函数，迭代结果和 X 的关联系数，使用的输入数据列，以及初始化窗口的长度，并最终在指定的输出列中输出结果。stateIterate 的计算规则参考文档：statelterate。实现代码如下：

trade = table(take("A", 6) join take("B", 6) as sym,  1..12 as val0,  take(10, 12) as val1)

inputTable = streamTable(1:0, `sym`val0`val1, [SYMBOL, INT, INT])
outputTable = table(100:0, `sym`factor, [STRING, DOUBLE])
engine = createReactiveStateEngine(
  name="rsTest",
  metrics=<[stateIterate(val0, val1, 3, msum{, 3}, [0.5, 0.5])]>,
  dummyTable=inputTable,
  outputTable=outputTable,
  keyColumn=["sym"],
  keepOrder=true)

engine.append!(trade)
select * from outputTable

上例中的 stateIterate(val0, val1, 3, msum{, 3}, [0.5, 0.5]) 基于 stateIterate 的计算规则，实现了线性递归，但仅从这一行代码无法理解其中的计算规则，对用户的使用可能造成困扰。如果通过 OOP 改写此函数的实现逻辑，则代码结构会清晰很多：

trade = table(take("A", 6) join take("B", 6) as sym,  1..12 as val0,  take(10, 12) as val1)

inputTable = streamTable(1:0, `sym`val0`val1, [SYMBOL, INT, INT])
outputTable = table(100:0, `sym`factor, [STRING, DOUBLE])

class MyIterateOperator {
	movingWindow :: DOUBLE VECTOR
	k :: INT
	def	MyIterateOperator() {
		k = 0
		movingWindow = double([])
	}
	def append(X, initial) {
		if (k < 3) {
			k = k + 1
			movingWindow = movingWindow join initial
			return double(initial)
		}
		result = 0.5 * X + 0.5 * sum(movingWindow)
		movingWindow = movingWindow[1:] join result
		return double(result)
	}
}
engine2 = createReactiveStateEngine(
  name="rsTest2",
  metrics=<[MyIterateOperator().append(val0, val1)]>,
  dummyTable=inputTable,
  outputTable=outputTable,
  keyColumn=["sym"],
  keepOrder=true)
  
engine2.append!(trade)
select * from outputTable

可以看到，当窗口长度小于3时，算子直接返回init中的结果；当窗口长度大于等于3时，算子的append方法将长度为3的窗口中数据的和与X中的值做加权平均；最终实现了和 stateIterate(val0, val1, 3, msum{, 3}, [0.5, 0.5]) 相同的功能。

虽然通过 OOP 实现线性递归的代码行数有所增加，但它结构清晰，提高了代码的可读性，同时简化了调试过程。