设置多个kafka connect sinks

Question

我正在努力将数据从 postgreSQL 流式传输到 HDFS。我在 HDP 2.6 沙箱上设置了融合环境。我的 jdbc postgreSQL 源配置是

name=jdbc_1
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
tasks.max=1
connection.url=jdbc:postgresql://host:port/db?currentSchema=schema&user=user&password=password
mode=timestamp
timestamp.column.name=col1
validate.non.null=false
topic.prefix=psql-

连接的所有其他属性也都很好，我运行通过

连接它

./bin/connect-standalone ./etc/kafka/connect-standalone.properties ./etc/kafka-connect-jdbc/source.properties

它工作正常并根据数据库中 table 的数量创建主题

psql-table1
psql-table2

现在我想运行 HDFS 吸收所有主题，为 postgreSQL 数据库中的每个 table 创建单独的目录。但是当我运行 HDFS 使用命令

下沉时

./bin/connect-standalone ./etc/kafka/connect-standalone.properties ./etc/kafka-connect-hdfs/hdfs-postGres.properties

来自运行我收到错误消息的来源

ERROR Stopping after connector error (org.apache.kafka.connect.cli.ConnectStandalone:113)
org.apache.kafka.connect.errors.ConnectException: Unable to start REST server
at org.apache.kafka.connect.runtime.rest.RestServer.start(RestServer.java:214)
at org.apache.kafka.connect.runtime.Connect.start(Connect.java:53)
at org.apache.kafka.connect.cli.ConnectStandalone.main(ConnectStandalone.java:95)
Caused by: java.net.BindException: Address already in use
at sun.nio.ch.Net.bind0(Native Method)
at sun.nio.ch.Net.bind(Net.java:433)
at sun.nio.ch.Net.bind(Net.java:425)
at sun.nio.ch.ServerSocketChannelImpl.bind(ServerSocketChannelImpl.java:223)
at sun.nio.ch.ServerSocketAdaptor.bind(ServerSocketAdaptor.java:74)
at org.eclipse.jetty.server.ServerConnector.openAcceptChannel(ServerConnector.java:331)
at org.eclipse.jetty.server.ServerConnector.open(ServerConnector.java:299)
at org.eclipse.jetty.server.AbstractNetworkConnector.doStart(AbstractNetworkConnector.java:80)
at org.eclipse.jetty.server.ServerConnector.doStart(ServerConnector.java:235)
at org.eclipse.jetty.util.component.AbstractLifeCycle.start(AbstractLifeCycle.java:68)
at org.eclipse.jetty.server.Server.doStart(Server.java:398)
at org.eclipse.jetty.util.component.AbstractLifeCycle.start(AbstractLifeCycle.java:68)
at org.apache.kafka.connect.runtime.rest.RestServer.start(RestServer.java:212)
... 2 more

如果我停止源连接并启动接收器，它就可以正常工作。任何人都可以帮助我如何设置多个接收器连接器。

Answer 1

Kafka Connect 在端口 8083 上启动一个休息服务器。

如果您在一台机器上运行多个独立连接器，则需要使用 rest.port 属性

进行更改

或者您可以运行 connect-distributed，然后 POST 您的源和接收器配置分别作为 JSON 负载运行在单个连接服务器上，那么你就不会有这个 Address already in use 问题了。

设置多个kafka connect sinks

Setup multiple kafka connect sinks

jdbc

apache-kafka

apache-kafka-connect

confluent-platform