//重复运行时清理环境
def clearEnv(){
	try{
	dropAggregator(`tsAggr1)
	dropAggregator(`tsAggr2)
	unsubscribeTable(,`level2,`act_tsAggr1)
	unsubscribeTable(,`level2,`act_tsAggr2)
	unsubscribeTable(,`level2,`act_factor)
	unsubscribeTable(,`level2,`newestLevel2data)
	undef(`level2,SHARED)
	undef(`OHLC1,SHARED)
	undef(`OHLC2,SHARED)
	undef(`FACTOR,SHARED)
	}catch(ex){}
	
}
clearEnv()

share streamTable(100:0, `symbol`datetime`last`askPrice1`bidPrice1`askVolume1`bidVolume1`volume, [SYMBOL,DATETIME,DOUBLE,DOUBLE,DOUBLE,INT,INT,INT]) as level2

dbPath = "dfs://level2Replay"
//使用回放功能将数据写入流表发布
dbDate = database("", VALUE, 2020.01.01..2020.12.31)
dbSymbol=database("", HASH, [SYMBOL, 10])
if(existsDatabase(dbPath)){
	dropDatabase(dbPath)
}
db = database(dbPath, COMPO, [dbDate, dbSymbol])
modal = table(1:0, `symbol`datetime`last`askPrice1`bidPrice1`askVolume1`bidVolume1`volume, [SYMBOL,DATETIME,DOUBLE,DOUBLE,DOUBLE,INT,INT,INT])
pt=db.createPartitionedTable(modal,`quotes, `datetime`symbol)
data = select symbol, datetime(datetime(date(date))+second(time)) as datetime, last, askPrice1, bidPrice1, askVolume1, bidVolume1, curVol as volume from loadTable("dfs://level2","quotes")
pt.append!(data)

quotes = loadTable(dbPath,"quotes")
//设置每次提取到内存数据量=1小时
repartitionSchema = time(cutPoints(08:00:00..18:00:00,10))
inputDS = replayDS(<select * from quotes>, `datetime, `datetime,  repartitionSchema)
submitJob("replay_quotes", "replay_quotes_stream",  replay,  [inputDS],  [`level2], `datetime, `datetime, 10, false, 2)