Kafka文件流连接和流API答案

【问题标题】：Kafka file stream connect and stream APIKafka文件流连接和流API
【发布时间】：2020-08-09 05:35:39
【问题描述】：

我正在处理文件流连接器，我在文件中有超过一千万条记录（它不是单个文件，它按帐户 # 分区）。我必须将这些文件加载到主题中并更新我的流。经历了独立的流，我有以下问题，需要帮助来实现。

查看数据集，我有两个帐户#，每个帐户有 5 行，我需要将它们分组为两行并作为 acctNbr 键。

如何编写我的源连接器来读取文件并获取分组逻辑？

我的代理在 Linux 机器 X、Y、Z 中运行。源连接器的后期开发，我的 jar 文件是否应该部署在每个代理中（如果我开始在分布式代理中运行）？
我只有 30 分钟的窗口来提取文件拖放到主题？有哪些参数可以调整逻辑以降低我的工作窗口？仅供参考，这个主题将有超过 50 个分区和 3 个代理设置。

数据集：

{"acctNbr":"1234567","secNbr":"AAPL","date":"2010-01-01","currentPrice":"10","availQnty":"10"}
{"acctNbr":"1234567","secNbr":"AAPL","date":"2010-01-02","currentPrice":"10","availQnty":"10"}
{"acctNbr":"1234567","secNbr":"AAPL","date":"2010-01-03","currentPrice":"10","availQnty":"10"}
{"acctNbr":"1234567","secNbr":"AAPL","date":"2010-01-04","currentPrice":"10","availQnty":"10"}
{"acctNbr":"1234567","secNbr":"AAPL","date":"2010-01-05","currentPrice":"10","availQnty":"10"}
{"acctNbr":"abc3355","secNbr":"AAPL","date":"2010-01-01","currentPrice":"10","availQnty":"10"}
{"acctNbr":"abc3355","secNbr":"AAPL","date":"2010-01-02","currentPrice":"10","availQnty":"10"}
{"acctNbr":"abc3355","secNbr":"AAPL","date":"2010-01-03","currentPrice":"10","availQnty":"10"}
{"acctNbr":"abc3355","secNbr":"AAPL","date":"2010-01-04","currentPrice":"10","availQnty":"10"}
{"acctNbr":"abc3355","secNbr":"AAPL","date":"2010-01-05","currentPrice":"10","availQnty":"10"}

【问题讨论】：

标签： apache-kafka apache-kafka-streams apache-kafka-connect kafka-producer-api

【解决方案1】：

如何编写我的源连接器来读取文件并获取分组逻辑

FileSream 连接器无法做到这一点，并且除了编写您自己的连接器的示例之外，它并没有用于此目的。换句话说，不要在生产中使用。

话虽如此，您可以使用 Flume、Filebeat、Fluentd、NiFi、Streamsets 等替代解决方案来glob您的文件路径，然后将所有记录逐行发送到卡夫卡主题。

源连接器的后期开发，我的 jar 文件应该部署在每个代理中

您不应在任何代理上运行 Connect。 Connect 服务器称为workers。

只有 30 分钟的窗口来提取文件拖放到主题？

不清楚这个数字是从哪里来的。上面列出的任何上述方法都监视所有新文件，没有任何定义的窗口。

【讨论】：