Kafka Connect FileStreamSource 忽略附加的行

Question

我正在开发一个使用Spark处理日志的应用程序，我想使用Kafka作为从日志文件中流式传输数据的一种方式。基本上，我有一个日志文件（在本地文件系统上），它会随着新日志不断更新，而Kafka Connect似乎是从文件中获取数据以及新附加行的完美解决方案。

我使用以下命令以默认配置启动服务器：

Zookeeper服务器：

< code > zookeeper-server-start . sh config/zookeeper . properties

动物园管理员属性

dataDir=/tmp/zookeeper
clientPort=2181
maxClientCnxns=0

Kafka服务器：

kafka-server-start.sh 配置/服务器属性

server.properties

broker.id=0
log.dirs=/tmp/kafka-logs
zookeeper.connect=localhost:2181
[...]

然后我创建了主题“连接测试”：

< code > Kafka-topics . sh-create-zookeeper localhost:2181-replication-factor 1-partitions 1-topic connect-test

最后我运行了Kafka连接器：

connect-standalone.sh配置/connect-standalone.properties配置/connect-file-source.properties

connect-standalone.properties（连接标准属性）

bootstrap.servers=localhost:9092
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=true

internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false

offset.storage.file.filename=/tmp/connect.offsets
offset.flush.interval.ms=10000

连接-文件-源.属性

name=my-file-connector
connector.class=FileStreamSource
tasks.max=1
file=/data/users/zamara/suivi_prod/app/data/logs.txt
topic=connect-test

首先，我通过运行一个简单的控制台消费者来测试连接器:

< code > Kafka-console-consumer . sh-bootstrap-server localhost:9092-topic connect-test-from-beginning

一切都很正常，消费者从文件中接收日志，当我添加日志时，消费者不断更新新的日志。

（然后，我按照本指南以“消费者”的身份尝试Spark：https://spark.apache.org/docs/2.2.0/streaming-kafka-0-8-integration.html#approach-2个直接接近的无接收器，它仍然很好）

在此之后，我从日志文件中删除了一些日志并更改了主题（我删除了“连接测试”主题，创建了另一个主题并使用新主题编辑了connect-file-source.properties）。

但是现在连接器不再以相同的方式工作。使用控制台使用者时，我只获取文件中已有的日志，并且忽略我添加的每个新行。也许在不更改连接器名称的情况下更改主题（和/或修改日志文件中的数据）会破坏 Kafka 中的某些内容......

这就是Kafka Connect对我的主题“新主题”和连接器“新文件连接器”所做的事情:

[2018-05-16 15:06:42,454] INFO Created connector new-file-connector (org.apache.kafka.connect.cli.ConnectStandalone:104)
[2018-05-16 15:06:42,487] INFO Cluster ID: qjm74WJOSomos3pakXb2hA (org.apache.kafka.clients.Metadata:265)
[2018-05-16 15:06:42,522] INFO Updated PartitionLeaderEpoch. New: {epoch:0, offset:0}, Current: {epoch:-1, offset:-1} for Partition: new-topic-0. Cache now contains 0 entries. (kafka.server.epoch.LeaderEpochFileCache)
[2018-05-16 15:06:52,453] INFO WorkerSourceTask{id=new-file-connector-0} Committing offsets (org.apache.kafka.connect.runtime.WorkerSourceTask:328)
[2018-05-16 15:06:52,453] INFO WorkerSourceTask{id=new-file-connector-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask:345)
[2018-05-16 15:06:52,458] INFO WorkerSourceTask{id=new-file-connector-0} Finished commitOffsets successfully in 5 ms (org.apache.kafka.connect.runtime.WorkerSourceTask:427)
[2018-05-16 15:07:02,459] INFO WorkerSourceTask{id=new-file-connector-0} Committing offsets (org.apache.kafka.connect.runtime.WorkerSourceTask:328)
[2018-05-16 15:07:02,459] INFO WorkerSourceTask{id=new-file-connector-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask:345)
[2018-05-16 15:07:12,459] INFO WorkerSourceTask{id=new-file-connector-0} Committing offsets (org.apache.kafka.connect.runtime.WorkerSourceTask:328)
[2018-05-16 15:07:12,460] INFO WorkerSourceTask{id=new-file-connector-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask:345)

（即使在文件中附加新行后，它也会不断刷新 0 条未完成的消息）

所以我尝试重新开始：我删除了/tmp/kafka日志目录/tmp/connect。偏移文件，并使用了全新的主题名称、连接器名称和日志文件，以防万一。但是，连接器仍然忽略了新的日志……我甚至试图删除我的kafka，从存档中重新提取它，然后再次运行整个过程（以防kafka中发生了变化），但没有成功。

我不知道问题出在哪里，任何帮助都将不胜感激！

Answer 1

Kafka Connect 不会“监视”或“跟踪”文件。我不相信它在任何地方都有记录表明它确实这样做了。

我会说它对于读取活动日志比使用 Spark Streaming 观看文件夹更不有用。Spark将“识别”新创建的文件。Kafka Connect FileStreamSource 必须指向单个预先存在的不可变文件。

要让 Spark 处理活动日志，您需要执行“日志轮换”的东西 - 也就是说，当文件达到最大大小或时间段（例如一天）结束等条件时，此过程会将活动日志移动到 Spark 正在监视的目录，然后它处理启动一个新的日志文件，以便您的应用程序继续写入。

如果您希望文件被主动监视并摄入到Kafka中，那么可以使用FileBeat、Fluentd或Apache Flume。

Answer 2

根据文档：

FileStream连接器示例旨在向那些作为用户或开发人员首次使用Kafka Connect的用户展示一个简单的连接器是如何运行的。不建议在生产中使用。

我会使用类似Filebeat（带有Kafka输出）的东西来将日志摄取到Kafka中。或者kafka connect spoldir，如果您的日志没有直接附加到，而是放在文件夹中的独立文件，以便接收。