我已正确配置了FilePulse,以便当我在阅读文件夹中创建文件时,它会在主题中读取并摄取该文件。
现在我需要连续阅读该文件夹中的每个文件,因为它们正在不断更新。
我必须更改属性文件的任何属性吗?
我的文件PulseTxtFile.properties:
name=connect-file-pulse-txt
connector.class=io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector
topic=lineas-fichero
tasks.max=1
File types
fs.scan.filters=io.streamthoughts.kafka.connect.filepulse.scanner.local.filter.RegexFileListFilter
file.filter.regex.pattern=.*\.log$
task.reader.class=io.streamthoughts.kafka.connect.filepulse.reader.RowFileInputReader
File scanning
fs.cleanup.policy.class=io.streamthoughts.kafka.connect.filepulse.clean.LogCleanupPolicy
fs.scanner.class=io.streamthoughts.kafka.connect.filepulse.scanner.local.LocalFSDirectoryWalker
fs.scan.directory.path=/home/ec2-user/parser/scanDirKafka
fs.scan.interval.ms=10000
Internal Reporting
internal.kafka.reporter.bootstrap.servers=localhost:9092
internal.kafka.reporter.id=connect-file-pulse-txt
internal.kafka.reporter.topic=connect-file-pulse-status
Track file by name
offset.strategy=name
非常感谢!
仅RowFileInputReader
支持连续读取,您可以使用read.max.wait.ms
属性配置--命中文件结尾后等待更多字节的最长时间(以毫秒为单位)。
例如,如果将该属性配置为10000
,则读取器将等待10秒以将新行添加到文件中,然后才会认为它已完成。
allow.tasks.reconfiguration.after.timeout.ms
配置为强制在给定时间段后重新启动所有任务,以便计划新文件。
最后,您必须注意正确设置max.tasks
属性,以便可以并行处理所有文件(一个任务一次只能处理一个文件)。
上一篇:有没有一种方法,使从内WebBrowser控件IE8的开发工具在.NET应用程序控件、开发工具、应用程序、方法
下一篇:如何获取Kafka Debezium MySQL连接器的数据库.server.name?连接器、数据库、Debezium、Kafka