时间戳

首页 > 解决方案 > Kafka Connect：JDBC Source Connector：创建具有多个分区的主题

jdbc - Kafka Connect：JDBC Source Connector：创建具有多个分区的主题

问题描述

我已经从 MySQL 创建了一个示例管道轮询数据并写入 HDFS（以及配置单元表）。

由于我的要求，我需要为每个数据库表创建 Source+Connector 对。下面我发布了我的源连接器和接收器连接器的配置设置。

我可以看到一个主题是使用一个分区创建的，并且复制因子为 1。

主题创建应该是自动的，这意味着我不能在创建 Source+Sink 对之前手动创建主题。

我的问题：

1）有没有办法在创建源连接器时配置分区数和复制因子？

2）如果可以创建多个分区，源连接器使用什么样的分区策略？

3) 应该为 Source 和 Sink 连接器创建多少正确的 worker？

源连接器：

{
  "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
  "mode": "timestamp+incrementing",
  "timestamp.column.name": "modified",
  "incrementing.column.name": "id",
  "topic.prefix": "jdbc_var_cols-",
  "tasks.max": "1",
  "poll.interval.ms": "1000",
  "query": "SELECT id,name,email,department,modified FROM test",
  "connection.url": "jdbc:mariadb://127.0.0.1:3306/connect_test?user=root&password=confluent"
}

水槽连接器：

{
  "connector.class": "io.confluent.connect.hdfs.HdfsSinkConnector",
  "topics.dir": "/user/datalake/topics-hive-var_cols3",
  "hadoop.conf.dir": "/tmp/quickstart/hadoop/conf",
  "flush.size": "5",
  "schema.compatibility": "BACKWARD",
  "connect.hdfs.principal": "datalake@MYREALM.LOCAL",
  "connect.hdfs.keytab": "/tmp/quickstart/datalake.keytab",
  "tasks.max": "3",
  "topics": "jdbc_var_cols-",
  "hdfs.url": "hdfs://mycluster:8020",
  "hive.database": "kafka_connect_db_var_cols3",
  "hdfs.authentication.kerberos": "true",
  "rotate.interval.ms": "1000",
  "hive.metastore.uris": "thrift://hive_server:9083",
  "hadoop.home": "/tmp/quickstart/hadoop",
  "logs.dir": "/logs",
  "format.class": "io.confluent.connect.hdfs.avro.AvroFormat",
  "hive.integration": "true",
  "hdfs.namenode.principal": "nn/_HOST@MYREALM.LOCAL",
  "hive.conf.dir": "/tmp/quickstart/hadoop/conf"
}

标签： jdbcapache-kafkahdfsapache-kafka-connectconfluent-platform

解决方案

1）有没有办法在创建源连接器时配置分区数和复制因子？

不是来自 Connect，不是。

听起来您在代理上启用了自动主题创建，因此它使用默认值。理想情况下，这应该在生产环境中禁用，因此您必须提前创建主题。

源连接器使用什么样的分区策略？

取决于哪个连接器以及代码的编写方式（即是否/如何生成记录的密钥）。例如，如果使用 JDBC 连接器，键可能是数据库表的主键。它将使用 DefaultPartitioner 进行散列。我不相信 Connect 允许您在每个连接器级别指定自定义分区器。如果键为空，则消息将分布在所有分区上。

3) 应该为 Source 和 Sink 连接器创建多少正确的 worker？

同样，取决于来源。对于 JDBC，每个表都有一个任务。

但是，对于接收器，任务只能达到被接收主题的分区数（与所有消费者组一样）。

此外，您通常会独立于数据库（和 Hadoop 集群）运行 Connect 集群

推荐阅读