为何Yarn显示MapReduce任务完成，JdbcSourceConnector仍在拉取Hive数据？

你遇到的问题核心是：Hive端SQL任务已完成，但JDBC Source Connector从Hive拉取并写入Kafka的耗时远高于Hive任务本身，主要原因集中在连接器的拉取/发送配置、JDBC驱动行为、并行度这几个方面，以下是具体排查点和解决办法：

1. JDBC结果集拉取效率过低 #

Hive JDBC驱动默认的fetchsize很小（通常为1000甚至更低），这意味着连接器每次从HiveServer2只能拉取少量数据。即使Hive端已经计算完所有结果，客户端也要反复多次请求才能把数据全部拉取完毕，大幅增加总耗时。

解决办法：
在JDBC连接URL中添加fetchsize参数，设置较大的值（比如10000或50000，根据单条数据大小调整）：

"connection.url": "jdbc:hive2://hive_host:10000;hive.resultset.use.unique.column.names=false;fetchsize=10000"

2. 连接器批量处理配置不合理 #

JdbcSourceConnector的默认批量参数偏保守，导致拉取数据和写入Kafka的频次过高，吞吐量上不去：

• batch.max.rows：默认值1000，即每次从数据库拉取1000条数据
• Kafka生产者的producer.batch.size：默认16KB，每次发送的批量数据太小

解决办法：
在连接器配置中添加以下参数，调大批量大小：

"batch.max.rows": "10000",
"producer.batch.size": "33554432",  // 32MB
"producer.linger.ms": "50"  // 等待50ms攒够数据再发送，提升批量效率

3. 连接器任务并行度不足 #

默认tasks.max=1，单线程处理所有数据的拉取和写入，即使Hive端是并行计算，连接器也无法利用多核或多线程提升速度。

解决办法：
如果你的Hive表是分区表，可以根据分区数设置合理的tasks.max值（比如分区数是8，就设为8），同时调整查询逻辑让每个任务处理一个分区，避免重复拉取：

"tasks.max": "4",
"query": "select * from your_table where dt = ?"  // 假设dt是分区字段，连接器会自动拆分任务

如果是非分区表，bulk模式下并行需要确保数据可以被分片（比如按主键范围拆分），可以结合mode=incrementing或timestamp模式；若必须用bulk，可手动拆分查询为多个子查询分配给不同任务。

4. Kafka生产者性能限制 #

默认的生产者配置可能未优化，导致写入Kafka的速度跟不上数据拉取速度：

• 未开启压缩：数据传输体积大，耗时久
• acks设置过高：默认acks=1，需要等待Broker确认后再发送下一批，增加延迟

解决办法：
添加生产者优化配置：

"producer.compression.type": "snappy",  // 开启snappy压缩，平衡压缩比和速度
"producer.acks": "1"  // 若对数据一致性要求不高，可临时设为0（不推荐生产环境），或保持1并调大其他参数

5. 序列化方式开销大 #

默认的JsonConverter序列化大体积数据时效率较低，二进制序列化格式（如Avro）的速度更快、体积更小。

解决办法：
换成Avro转换器（需Confluent Schema Registry支持）：

"value.converter": "io.confluent.connect.avro.AvroConverter",
"value.converter.schema.registry.url": "http://your-schema-registry:8081"

内容的提问来源于stack exchange，提问作者X.cheer