csv - 将 CSV 文件中的 DateTime 字符串解析为事件时间戳

Question

使用 WSO2 SP，我的应用程序从以下 CSV 文件中读取行：

20170801 000001237,1.321420,1.321510,0
20170801 000001487,1.321440,1.321530,0
20170801 000001737,1.321450,1.321530,0
20170801 000001987,1.321440,1.321530,0

第一列是时间戳字符串，需要对其进行解析以对事件进行时间戳记：

yyyyMMdd hhmmssfff

, 其中 fff 是毫秒

我当前的应用程序如下所示：

@App:name('ReceiveAndCount')
@App:description('count events in csv file')
@source(type = 'file', 
    mode='line',
    tailing='false',
    file.uri = "file:/Users/A/Desktop/siddhi/wso2sp-4.4.0/data/DAT_ASCII_GBPUSD_T_201708.csv", 
    action.after.process='NONE',
        @map(type = 'csv', header='false', delimiter = ",",
        @attributes(
        dateTime = '0',
        bid='1',
        ask='2',
        ignore='3'  ) ))

define stream csvGBPUSDstream (dateTime string, bid double, ask double, ignore int);

@sink(type = 'log', priority='info')
define stream TotalCountStream (totalCount long);

-- Count the incoming events
@info(name = 'query1')
from csvGBPUSDstream 
select count() as totalCount
insert into TotalCountStream;

非常感谢有关将第一列中的日期字符串解析为事件时间戳的任何指导

标签： csvtimestampetlwso2sp

Answer 1

通过以下命令在 docker 中运行 wso2sp-editor，使用音量开关允许读取本地数据文件：

docker run -it -p 9390:9390 -v /Users/A/Desktop/wso2sp-editor-docker-data:/home/wso2carbon/wso2sp-4.4.0/data/ --name editor wso2/wso2sp-editor:4.4.0

最后是应用代码：

 @App:name('CsvParseDateTime')

@App:description('parse timestamp string from csv file') --abrakadabra @source(type = 'file', mode = "line", tailing = "false", file.uri = "file://home/wso2carbon /wso2sp-4.4.0/data/DAT_ASCII_GBPUSD_T_201708.csv", --"file:/Users/A/Desktop/siddhi/wso2sp-4.4.0/data/DAT_ASCII_GBPUSD_T_201708.csv", action.after.process = "NONE" , @map(type = 'csv', header = "false", delimiter = ",", @attributes(dateTime = "0", ask = "2", ignore = "3", bid = "1")) ) 定义流 csvGBPUSDstream (dateTime 字符串，bid double，ask double，忽略 int)；

@sink(type = 'log', priority = "info") 定义流 FiveSecRangeStream (timestamp long, FiveSecRange double);

-- streat 将 dateTime 字符串解析为可以用作时间窗口索引的毫秒时间 --time:timestampInMilliseconds(date.value,date.format) yyyy-MM-dd HH:mm:ss.SSS @info(name = 'Parse Timestamp') from csvGBPUSDstream select time:timestampInMilliseconds(dateTime, 'yyyyMMdd HHmmssSSS') 作为时间戳，出价，询价插入 indexedGBPUSDstream；

-- 外部窗口时间（事件流中的时间戳）来自 indexedGBPUSDstream#window.externalTime(timestamp,5 Sec) 选择时间戳，max(bid)-min(bid) as FiveSecRange 插入到 FiveSecRangeStream 中；-- 适用于所有事件；