如何通过编程方式将大量JSON文件推送至Kafka Topic？

嘿，这个需求我之前帮不少朋友解决过，20万条JSON文件推送到Kafka、每个文件作为单条消息的场景，其实有好几种靠谱的实现方案，我给你拆解清楚，你可以根据自己的技术栈和实际场景来选：

如果你只是想快速完成推送，不想写代码，用Kafka自带的kafka-console-producer.sh配合shell脚本就能搞定。核心思路是遍历所有JSON文件，把每个文件的完整内容作为一条消息发送。

示例shell脚本：

#!/bin/bash
# 替换成你的Kafka Topic名称
KAFKA_TOPIC="your_target_topic"
# 替换成你的Kafka Broker地址
KAFKA_BROKER="localhost:9092"
# 替换成你的JSON文件存放目录
JSON_DIR="/path/to/json/files"

# 遍历目录下所有JSON文件（含子目录的话用find命令）
find "$JSON_DIR" -type f -name "*.json" | while read file; do
  if [ -f "$file" ]; then
    # 将文件内容作为单条消息发送到Kafka
    cat "$file" | kafka-console-producer.sh --broker-list "$KAFKA_BROKER" --topic "$KAFKA_TOPIC"
    # 可选：打印处理日志，方便追踪进度
    echo "已推送文件: $file"
  fi
done

注意事项：

• 如果想提高推送速度，可以给producer加参数调整批量策略，比如--batch-size 16384 --linger-ms 5，但要注意linger-ms别设太高，不然会把多个文件的内容攒成一批发送，破坏“每个文件一条消息”的要求。
• 可以把推送失败的文件路径记录到日志，方便后续重发。

如果需要加自定义逻辑（比如校验JSON格式、添加业务字段、失败重试），用代码实现会更灵活。这里给你举Python和Java两个最常用的例子：

Python 实现（用kafka-python库） #

先安装依赖：pip install kafka-python

示例代码：

from kafka import KafkaProducer
import os
import json

def push_json_files():
    # 初始化Kafka Producer
    producer = KafkaProducer(
        bootstrap_servers="localhost:9092",
        # 如果文件内容是已格式化的JSON字符串，直接encode即可，不用json.dumps
        value_serializer=lambda x: json.dumps(x).encode('utf-8')
    )
    topic = "your_target_topic"
    json_dir = "/path/to/json/files"

    # 遍历所有JSON文件（含子目录）
    for root, _, files in os.walk(json_dir):
        for file in files:
            if file.endswith(".json"):
                file_path = os.path.join(root, file)
                try:
                    with open(file_path, "r", encoding="utf-8") as f:
                        # 读取文件内容（如果是单个JSON对象，直接load；如果是纯字符串，用f.read()）
                        content = json.load(f)
                        # 发送消息
                        producer.send(topic, value=content)
                        print(f"已推送: {file_path}")
                except Exception as e:
                    print(f"处理失败 {file_path}: {str(e)}")
                    # 可以把失败文件写入日志文件，后续重发
                    with open("failed_files.log", "a") as log:
                        log.write(f"{file_path}\n")
    # 确保所有消息都被发送
    producer.flush()
    producer.close()

if __name__ == "__main__":
    push_json_files()

优化技巧：可以用concurrent.futures.ThreadPoolExecutor开启多线程并行推送，提高速度（Kafka Producer是线程安全的，共用一个实例即可）。

Java 实现（用官方Kafka Client） #

先添加Maven依赖：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.5.1</version> <!-- 版本请匹配你的Kafka集群 -->
</dependency>

示例代码：

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.util.Properties;

public class JsonFileKafkaProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        String topic = "your_target_topic";
        File jsonDir = new File("/path/to/json/files");

        processDirectory(jsonDir, producer, topic);

        producer.flush();
        producer.close();
    }

    private static void processDirectory(File dir, KafkaProducer<String, String> producer, String topic) {
        File[] files = dir.listFiles();
        if (files == null) return;

        for (File file : files) {
            if (file.isDirectory()) {
                processDirectory(file, producer, topic);
            } else if (file.getName().endsWith(".json")) {
                try {
                    // 读取文件完整内容作为消息体
                    String content = Files.readString(file.toPath());
                    ProducerRecord<String, String> record = new ProducerRecord<>(topic, content);
                    // 异步发送并处理回调
                    producer.send(record, (metadata, exception) -> {
                        if (exception != null) {
                            System.err.println("推送失败 " + file.getAbsolutePath() + ": " + exception.getMessage());
                        } else {
                            System.out.println("已推送 " + file.getAbsolutePath());
                        }
                    });
                } catch (IOException e) {
                    System.err.println("读取文件失败 " + file.getAbsolutePath() + ": " + e.getMessage());
                }
            }
        }
    }
}

如果这是一个需要长期运行的数据流任务，或者你不想维护代码，Kafka Connect是最佳选择。可以用第三方的FilePulseSourceConnector，它支持把整个文件作为一条消息发送。

步骤：

1. 下载FilePulseSourceConnector的jar包，放到Kafka Connect的插件目录（比如$KAFKA_HOME/plugins）。
2. 创建连接器配置文件file-pulse-json-source.properties：

name=json-file-source-connector
connector.class=io.streamsets.pipeline.stage.source.file.FilePulseSourceConnector
tasks.max=4 # 多任务并行处理，提高推送速度
fs.scan.directory.path=/path/to/json/files
fs.scan.interval.ms=1000
fs.file.pattern=*.json
fs.read.mode=READ_ONCE # 避免重复推送已处理的文件
producer.override.bootstrap.servers=localhost:9092
producer.override.topic=your_target_topic
record.type=WHOLE_FILE # 核心配置：每个文件作为一条完整消息

3. 启动Kafka Connect分布式模式，然后提交连接器配置：

# 启动Connect
$KAFKA_HOME/bin/connect-distributed.sh $KAFKA_HOME/config/connect-distributed.properties

# 提交连接器配置
curl -X POST -H "Content-Type: application/json" --data @file-pulse-json-source.properties http://localhost:8083/connectors

• 性能优化：20万文件量不算小，建议开启多线程/多任务并行处理，同时调整Kafka Producer的batch.size、buffer.memory等参数，但要保证linger.ms不要过高，避免多个文件内容被合并成一条消息。
• 重复推送防护：如果任务中途中断，要避免重复发送已成功的文件。可以用“处理完文件就移到归档目录”的方式，或者在代码里记录已处理的文件名，Connect的READ_ONCE模式也会自动标记已处理文件。
• 错误处理：一定要记录失败的文件路径，方便后续手动重发或者自动重试。

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