如何使用Kafka Source和Sink连接器实现Hive与MySQL的连接？

MySQL 到 Kafka（Source 连接器） #

Confluent 的 JDBC Source Connector 完全支持 MySQL 作为数据源，以下是可用的配置示例：

name=mysql-source
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
tasks.max=1
connection.url=jdbc:mysql://<mysql-host>:3306/<db-name>?user=<username>&password=<password>
mode=incrementing
incrementing.column.name=<主键自增列>
topic.prefix=mysql-
table.whitelist=<需要同步的表名>

• mode 可选 incrementing（基于自增主键增量同步）或 timestamp（基于时间戳字段），根据业务场景选择
• 配置完成后，指定的 MySQL 表数据会同步到前缀为 mysql- 的 Kafka Topic 中

Kafka 到 Hive（Sink 连接器） #

官方没有直接的 Hive Sink Connector，但有两种稳定可行的实现方式：

方式1：通过 JDBC Sink 直接连接 Hive #

Hive 提供 JDBC 驱动，可复用 Confluent 的 JDBC Sink Connector 实现数据写入，配置示例：

name=hive-jdbc-sink
connector.class=io.confluent.connect.jdbc.JdbcSinkConnector
tasks.max=1
connection.url=jdbc:hive2://<hive-host>:10000/<hive-db-name>
connection.user=<hive-username>
connection.password=<hive-password>
topics=<对应MySQL的Kafka Topic名>
auto.create=true
auto.evolve=true
insert.mode=upsert
pk.fields=<主键字段>

注意：需要将 Hive 的 JDBC 驱动包（如 hive-jdbc-<version>.jar）放置到 Kafka Connect 的插件目录，重启 Connect 服务后生效。

方式2：先写入 HDFS 再映射 Hive 外部表 #

适合大数据量批量同步场景，性能更优：

1. 用 HDFS Sink Connector 将 Kafka 数据写入 HDFS：

name=hdfs-sink
connector.class=io.confluent.connect.hdfs.HdfsSinkConnector
tasks.max=1
topics=<对应MySQL的Kafka Topic名>
hdfs.url=hdfs://<namenode-host>:9000/<hive-data-path>
flush.size=1000
rotate.interval.ms=3600000
format.class=io.confluent.connect.hdfs.parquet.ParquetFormat

2. 在 Hive 中创建外部表映射该路径：

CREATE EXTERNAL TABLE <hive-table-name> (
  id INT,
  col1 STRING,
  col2 BIGINT
  -- 按实际表结构定义字段
)
STORED AS PARQUET
LOCATION 'hdfs://<namenode-host>:9000/<hive-data-path>/<topic-name>/';

Hive 到 MySQL 反向同步 #

如果需要从 Hive 同步数据到 MySQL，可通过以下流程实现：

• 用 Hive EXPORT 命令将数据导出到 HDFS
• 用 HDFS Source Connector 将 HDFS 数据读取到 Kafka Topic
• 再通过 JDBC Sink Connector 将 Kafka 数据写入 MySQL

内容的提问来源于stack exchange，提问作者Siddharth