如何配置Kafka-CONNECT-FILE-PULSE以连续读取文本文件？ [英] How can be configured kafka-connect-file-pulse for continuous reading of a text file?

问题描述

我已正确配置了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属性，以便可以并行处理所有文件(一个任务一次只能处理一个文件)。

这篇关于如何配置Kafka-CONNECT-FILE-PULSE以连续读取文本文件？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！