CentOS HDFS与Kafka集成应用案例

发布时间:2025-06-02 11:59:38

阅读量:17

在CentOS上集成Hadoop分布式文件系统（HDFS）与Apache Kafka，通常涉及将Kafka作为数据的生产者或消费者，并将数据写入HDFS或从HDFS读取数据。以下是一个简化的应用案例，展示如何使用Kafka将数据写入HDFS。

应用案例：使用Kafka将数据写入HDFS

环境准备：

确保CentOS系统上已安装Hadoop和Kafka。
配置好Kafka的生产者和消费者。

Kafka生产者配置：

创建一个Kafka主题，用于生产数据。

kafka-topics.sh --create --topic order-created-topic --partitions 12 --replication-factor 3 --bootstrap-server localhost:9092

编写Kafka生产者代码：

使用Kafka Producer API将数据发送到Kafka主题。

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

Producer producer = new KafkaProducer<>(props);

producer.send(new ProducerRecord<>("order-created-topic", orderId, orderJson));
producer.close();

将数据写入HDFS：

在Kafka消费者中读取数据，并将数据写入HDFS。可以使用Spark Streaming等框架来实现实时数据处理和写入。

SparkConf conf = new SparkConf().setAppName("Kafka to HDFS");
conf.setMaster("local[*]");

JavaPairRDD lines = KafkaUtils.createDirectStream(
    conf,
    "order-created-topic",
    new StringDeserializer(),
    new StringDeserializer()
).mapToPair(record -> new Tuple2<>(record.value(), record.key()));

lines.saveAsHadoopFile("/path/to/hdfs/directory",
    new TextOutputFormat(),
    "org.apache.hadoop.mapred.lib.MultipleTextOutputFormat",
    new Configuration(false)
);