请问file source 只读一次,最大读取的行数是多少?现在文件中有10000行,实际发出数据只有5000行,可以调整上限吗?
并没有限制。麻烦贴一下版本,规则及其相关配置(导出规则可得)以及重现步骤
this is rule data
{
“streams”: {},
“tables”: {},
“rules”: {
“0b519b50765511ef9f1e13f91c340ddf”: “{"triggered":false,"id":"0b519b50765511ef9f1e13f91c340ddf","name":"提取高度速度1","graph":{"nodes":{"file6860":{"type":"source","nodeType":"file","props":{"actionAfterRead":0,"columns":null,"confKey":"default","datasource":"86.status","fileType":"json","format":"json","hasHeader":false,"ignoreEndLines":0,"ignoreStartLines":0,"interval":0,"moveTo":"","node":{"class":"BasicNode, is-source","data":{"dataSource":{"default":"test.json","hint":{"en_US":"The file name without directory, e.g. test.json","zh_CN":"不含路径的文件名,例如 test.json"},"label":{"en_US":"Data Source (File or directory relative path)","zh_CN":"数据源(文件或者目录的相对地址)"}},"group":"source","label":"File","name":"file","numValue":6860,"properties":[{"connection_related":false,"control":"select","default":"json","exist":false,"hint":{"en":"The file format type.","zh":"文件格式类型"},"label":{"en":"File type","zh":"文件类型"},"name":"fileType","optional":false,"type":"string","values":["json","csv","lines"]},{"connection_related":false,"control":"text","default":"","exist":false,"hint":{"en":"The directory of the file relative to eKuiper root or an absolute path. Do not include the file name here. The file name should be defined in the stream data source.","zh":"指定文件的目录,可以是软件根目录的相对路径,或绝对路径。请勿包含文件名,文件名应该在数据源中定义。"},"label":{"en":"Path","zh":"文件夹路径"},"name":"path","optional":false,"type":"string","values":null},{"connection_related":false,"control":"text","default":0,"exist":false,"hint":{"en":"The interval between reading the files, time unit is ms. If only read once, set it to 0","zh":"读取文件的间隔时间,单位为毫秒。如果只需读取一次,设置为 0。"},"label":{"en":"Interval","zh":"间隔时间"},"name":"interval","optional":true,"type":"int","values":null},{"connection_related":false,"control":"text","default":0,"exist":false,"hint":{"en":"The sending interval between each event in millisecond.","zh":"事件发送的间隔时间,单位为毫秒。"},"label":{"en":"Send Interval","zh":"发送间隔"},"name":"sendInterval","optional":true,"type":"int","values":null},{"connection_related":false,"control":"select","default":0,"exist":false,"hint":{"en":"The action after read. 0 meas keep the file; 1 means delete the file; 2 means move the file to the path defined in the property ‘moveTo’","zh":"读取后的操作。0 表示保留文件;1 表示删除文件;2 表示将文件移动到属性 ‘移动位置’ 中定义的路径。"},"label":{"en":"Action after read","zh":"读取后动作"},"name":"actionAfterRead","optional":true,"type":"int","values":[0,1,2]},{"connection_related":false,"control":"text","default":"","exist":false,"hint":{"en":"The path to move the file to after read, only valid when the actionAfterRead is 1.","zh":"移动文件的位置, 仅用于 actionAfterRead 为 2 的情况"},"label":{"en":"Move to path","zh":"移动位置"},"name":"moveTo","optional":true,"type":"string","values":null},{"connection_related":false,"control":"radio","default":false,"exist":false,"hint":{"en":"If the first line is header, usually used for csv file.","zh":"是否包含文件头,多用于 csv。若为 true,则第一行解析为文件头。"},"label":{"en":"Has header","zh":"是否包含文件头"},"name":"hasHeader","optional":true,"type":"bool","values":null},{"connection_related":false,"control":"list","default":null,"exist":false,"hint":{"en":"Define the columns. If header is defined, this will be override.","zh":"定义文件的列。如果定义了文件头,该选项将被覆盖。"},"label":{"en":"Columns","zh":"字段列表"},"name":"columns","optional":true,"type":"list_string","values":null},{"connection_related":false,"control":"text","default":0,"exist":false,"hint":{"en":"How many lines to be ignored at the beginning. Notice that, empty line will be ignored and not be calculated.","zh":"忽略开头多少行的内容。"},"label":{"en":"Ignore start lines","zh":"文件开头忽略的行数"},"name":"ignoreStartLines","optional":true,"type":"int","values":null},{"connection_related":false,"control":"text","default":0,"exist":false,"hint":{"en":"How many lines to be ignored in the end. Notice that, empty line will be ignored and not be calculated.","zh":"忽略结尾多少行的内容。最后的空行不计算在内。。"},"label":{"en":"Ignore end lines","zh":"文件结尾忽略的行数"},"name":"ignoreEndLines","optional":true,"type":"int","values":null}],"record":{"actionAfterRead":0,"columns":null,"confKey":"default","datasource":"86.status","fileType":"json","format":"json","hasHeader":false,"ignoreEndLines":0,"ignoreStartLines":0,"interval":0,"moveTo":"","path":"","sendInterval":0},"schemaDesc":"监控文件系统,读取文件内容,并且将数据放入 eKuiper 数据处理流水线中。","type":"output"},"html":"BasicNode","id":1,"inputs":{},"name":"file","outputs":{"output_1":{"connections":[{"node":"4","output":"input_1"}]}},"pos_x":133,"pos_y":243,"typenode":"vue"},"path":"","sendInterval":0},"ui":null},"script2354":{"type":"operator","nodeType":"script","props":{"node":{"class":"BasicNode, is-operator","data":{"dataSource":,"group":"operator","label":"脚本","name":"script","numValue":2354,"properties":[{"control":"textarea","default":"","hint":{"en_US":"The script text, now supports JavaScript only. Must have a function named exec, whose parameter is the message and the return value must be the processed message.","zh_CN":"脚本函数的文本,支持javascript。必须包含一个函数,函数名为exec,参数为message,返回值为处理后的message"},"label":{"en_US":"Script","zh_CN":"脚本"},"name":"script","optional":false,"type":"string"}],"record":{"script":"/\n * 提取高度速度\n * */\n function exec(message)\n {\n if (message[‘data’]) {\n if (message[‘data’]?.[‘Type’] === ‘0’) {\n let data = {\n \"table1\": {\n \"sd\": parseInt(message[‘data’][‘vel’]),\n \"gd\": parseInt(message[‘data’][‘hei’])\n }\n };\n return {\n \"dataset_id\": \"dcce047a-8351-42b0-8e8a-87f54f6366c7\",\n \"deduction_id\": message[‘UUID’],\n \"timestamp\": message[‘engineCurtime’],\n \"data\": JSON.stringify(data)\n }\n }\n }\n return {\n \"dataset_id\": \"dcce047a-8351-42b0-8e8a-87f54f6366c7\",\n \"deduction_id\":0,\n \"data\": null,\n \"timestamp\": 0\n };\n }"},"schemaDesc":"用于针对数据运行脚本的操作","type":"default"},"html":"BasicNode","id":4,"inputs":{"input_1":{"connections":[{"input":"output_1","node":"1"}]}},"name":"script","outputs":{"output_1":{"connections":[{"node":"10","output":"input_1"}]}},"pos_x":383,"pos_y":388,"typenode":"vue"},"script":"/\n * 提取高度速度\n * */\n function exec(message)\n {\n if (message[‘data’]) {\n if (message[‘data’]?.[‘Type’] === ‘0’) {\n let data = {\n \"table1\": {\n \"sd\": parseInt(message[‘data’][‘vel’]),\n \"gd\": parseInt(message[‘data’][‘hei’])\n }\n };\n return {\n \"dataset_id\": \"dcce047a-8351-42b0-8e8a-87f54f6366c7\",\n \"deduction_id\": message[‘UUID’],\n \"timestamp\": message[‘engineCurtime’],\n \"data\": JSON.stringify(data)\n }\n }\n }\n return {\n \"dataset_id\": \"dcce047a-8351-42b0-8e8a-87f54f6366c7\",\n \"deduction_id\":0,\n \"data\": null,\n \"timestamp\": 0\n };\n }"},"ui":null},"sql1786":{"type":"sink","nodeType":"sql","props":{"confKey":"","fields":["dataset_id","deduction_id","data","timestamp"],"format":"json","name":251,"node":{"class":"BasicNode, is-sink","data":{"dataSource":,"group":"sink","label":"SQL","name":"sql","numValue":1786,"properties":[{"connection_related":false,"control":"text","default":"","exist":false,"hint":{"en":"The url of the database, it is important to note that the password in the URL is url-encoded characters, and you need to re-enter the password when editing. For example, the database address of mysql is mysql://username:password@127.0.0.1:3306/testdb?parseTime=true, if the original password is 123:#?, the password here should be filled in with 123%3A%23%3F","zh":"数据库服务器的 URL,需要特别注意的是 URL 中的密码为 url 编码后的字符,并且编辑时需要重新输入密码。例如,mysql 的数据库地址为 mysql://username:password@127.0.0.1:3306/testdb?parseTime=true,若原始密码为 123:#?,则这里的 password 应填入 123%3A%23%3F"},"label":{"en":"server address","zh":"数据库地址"},"name":"url","optional":false,"type":"string","values":null},{"connection_related":false,"control":"text","default":"","exist":false,"hint":{"en":"Table name","zh":"表名"},"label":{"en":"Table name","zh":"表名"},"name":"table","optional":false,"type":"string","values":null},{"connection_related":false,"control":"list","default":null,"exist":false,"hint":{"en":"Tag fields written to the database","zh":"写入数据库的标签字段"},"label":{"en":"Tag Fields","zh":"标签字段"},"name":"fields","optional":true,"type":"list_string","values":null},{"connection_related":false,"control":"radio","default":"","exist":false,"hint":{"en":"The output messages are received as an array. This is indicate whether to send the results one by one. If false, the output message will be {\"result\":\"${the string of received message}\"}. For example, {\"result\":\"[{\"count\":30},\"\"count\":20}]\"}. Otherwise, the result message will be sent one by one with the actual field name. For the same example as above, it will send {\"count\":30}, then send {\"count\":20} to the RESTful endpoint.Default to false.","zh":"如果选择结果为空,则忽略输出。"},"label":{"en":"Send single","zh":"是否忽略输出"},"name":"omitIfEmpty","optional":true,"type":"bool","values":},{"connection_related":false,"control":"radio","default":"","exist":false,"hint":{"en":"If the configuration item is set to true, when SELECT result is empty, then the result will not feed to sink operator.","zh":"输出消息以数组形式接收,该属性意味着是否将结果一一发送。 如果为 false,则输出消息将为{\"result\":\"${the string of received message}\"}。 例如,{\"result\":\"[{\"count\":30},\"\"count\":20}]\"}。否则,结果消息将与实际字段名称一一对应发送。 对于与上述相同的示例,它将发送 {\"count\":30},然后发送{\"count\":20} 到 RESTful 端点。默认为 false。"},"label":{"en":"Omit if content is empty","zh":"将结果数据按条发送"},"name":"sendSingle","optional":true,"type":"bool","values":},{"connection_related":false,"control":"select","default":"","exist":false,"hint":{"en":"","zh":""},"label":{"en":"Stream Format","zh":"流格式"},"name":"format","optional":true,"type":"string","values":["json","binary","protobuf","delimited","custom"]},{"connection_related":false,"control":"select","default":"","exist":true,"hint":{"en":"","zh":""},"label":{"en":"Schema Name","zh":"模式名称"},"name":"schemaName","optional":true,"type":"string","values":},{"connection_related":false,"control":"text","default":"","exist":true,"hint":{"en":"","zh":""},"label":{"en":"Schema Message","zh":"模式消息"},"name":"schemaMessage","optional":false,"placeholder":"","type":"string","values":null},{"connection_related":false,"control":"text","default":",","exist":true,"hint":{"en":"","zh":""},"label":{"en":"Delimited","zh":"分隔符"},"name":"delimiter","optional":false,"placeholder":"","type":"string","values":null},{"connection_related":false,"control":"textarea","default":"","exist":false,"hint":{"en":"The golang template format string to specify the output data format. The input of the template is the sink message which is always an array of map. If no data template is specified, the raw input will be the data.","zh":"Golang 模板格式字符串,用于指定输出数据格式。 模板的输入是目标消息,该消息始终是 map 数组。 如果未指定数据模板,则将数据作为原始输入。"},"label":{"en":"Data template","zh":"数据模版"},"name":"dataTemplate","optional":false,"type":"string","values":null}],"record":{"confKey":"","fields":["dataset_id","deduction_id","data","timestamp"],"format":"json","name":251,"sendSingle":true,"table":"eva_ekuiper_data_r","url":"pgsql://postgres:123456@10.8.8.30:5432/eva-agent"},"schemaDesc":"本插件为 SQL Database 的持久化插件,可以用于将分析数据存入支持 SQL 语法的数据库中","type":"input"},"html":"BasicNode","id":10,"inputs":{"input_1":{"connections":[{"input":"output_1","node":"4"}]}},"name":"sql","outputs":{},"pos_x":538,"pos_y":601,"typenode":"vue"},"sendSingle":true,"table":"eva_ekuiper_data_r","url":"pgsql://postgres:123456@10.8.8.30:5432/eva-agent"},"ui":null}},"topo":{"sources":["file6860"],"edges":{"file6860":["script2354"],"script2354":["sql1786"]}}},"options":{"debug":false,"logFilename":"","isEventTime":false,"lateTolerance":1000,"concurrency":1,"bufferLength":1024,"sendMetaToSink":false,"sendError":true,"qos":0,"checkpointInterval":300000,"restartStrategy":{"attempts":0,"delay":1000,"multiplier":2,"maxDelay":30000,"jitterFactor":0.1},"cron":"","duration":"","cronDatetimeRange":null}}”
},
“nativePlugins”: {
“sinks_sql”: “{"name":"sql","file":"file:///kuiper/data/uploads/sql_amd64.zip","shellParas":}”
},
“portablePlugins”: {},
“sourceConfig”: {
“file”: “{"default":{"actionAfterRead":0,"fileType":"lines","hasHeader":false,"ignoreEndLines":0,"ignoreStartLines":0,"interval":0,"moveTo":"/tmp/kuiper/moved","name":"default","parallel":false,"path":"/kuiper/data/uploads/","sendInterval":100}}”
},
“sinkConfig”: {},
“connectionConfig”: {},
“Service”: {},
“Schema”: {},
“uploads”: {},
“scripts”: {}
}