如何运行mongo-kafka连接器作为kafka的源并将其与logstash输入集成以使用elasticsearch作为接收器？

Question

我创建了https://github.com/mongodb/mongo-kafka的构建

但是这个运行如何与我正在运行的kafka实例连接。

甚至这个问题听起来多么愚蠢。但似乎没有任何文档可以使其与MongoDB的本地运行的Replicaset一起工作。

如果你有一个好的资源，请引导我走向它。

更新1--

已使用maven插件-https://search.maven.org/artifact/org.mongodb.kafka/mongo-kafka-connect

更新2--如何启用mongodb作为Kafka的源代码？

https://github.com/mongodb/mongo-kafka/blob/master/config/mongoSourceConnector.properties

要用作Kafka配置的文件

bin/kafka-server-start.sh config/server.properties --override config/MongoSourceConnector.properties

安装了汇流式和汇流式集线器，但仍不确定与Kafka一起工作的mongo-connector。

更新4-

Zookeeper,Kafka服务器,Kafka connect正在运行

bin/zookeeper-server-start.sh config/zookeeper.properties
bin/kafka-server-start.sh config/server.properties
bin/connect-standalone.sh config/connect-standalone.properties config/MongoSourceConnector.properties

使用logstash的以下配置，我可以将数据推入elasticsearch-

input {
  kafka {
        bootstrap_servers => "localhost:9092"
        topics => ["users","organisations","skills"]
  }
}
output {
  elasticsearch {
        hosts => ["localhost:9200"]
  }
  stdout { codec => rubydebug }
}

因此，现在一个MongoSourceConnector.properties保持一个它读取的集合名称，我需要为每个集合运行不同的属性文件的kafka connect。

我的Logstash正在将新数据推入elasticsearch，而不是更新旧数据。此外，它不是根据集合的名称创建索引。想法是这应该能够与我的MongoDB数据库完美地同步。

最终更新-现在一切工作顺利,

• 为kafka Connect创建了多个属性文件
• 最新的日志存储实际上根据主题名称创建索引，并相应地更新索引

input {
    kafka {
        bootstrap_servers => "localhost:9092"
        decorate_events => true
        topics => ["users","organisations","skills"]
    }
}
filter {
    json {
        source => "message"
        target => "json_payload"
    }

    json {
        source => "[json_payload][payload]"
        target => "payload"
    }

    mutate {
        add_field => { "[es_index]" => "%{[@metadata][kafka][topic]}" }
        rename => { "[payload][fullDocument][_id][$oid]" => "mongo_id"}
        rename => { "[payload][fullDocument]" => "document"}
        remove_field => ["message","json_payload","payload"]
    }
}
output {
    elasticsearch {
        hosts => ["localhost:9200"]
        index => "%{es_index}"
        action => "update"
        doc_as_upsert => true
        document_id => "%{mongo_id}"
    }
    stdout {
        codec =>
        rubydebug {
            metadata => true
        }
    }
}

Answer 1

端口8083是Kafka Connect，您可以从connect-*.sh脚本之一开始。

它是独立于代理的，属性不是从kafka-server-start中设置的