CentOS HDFS与Kafka集成应用案例-CentOS-PHP中文网

CentOS HDFS与Kafka集成应用案例

煙雲

发布： 2025-07-19 08:32:28

原创

914人浏览过

在CentOS系统上整合Hadoop分布式文件系统（HDFS）与Apache Kafka，通常会把Kafka当作数据的生成器或接收器，并且将数据存储到HDFS或者从HDFS获取数据。下面是一个简化版的应用场景，演示了怎样利用Kafka把数据存储到HDFS。

场景：利用Kafka向HDFS存入数据

初始化设置：

确认CentOS里已经装好了Hadoop和Kafka。
对Kafka的生成器和接收器做好配置。

Kafka生成器配置：

建立一个Kafka主题，用来生成数据。

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

登录后复制

构建Kafka生成器代码：

利用Kafka Producer API把数据传送到Kafka主题。

<code>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<string string=""> producer = new KafkaProducer(props);

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

登录后复制

数据导入HDFS：

在Kafka接收器中读取数据，并且把数据写进HDFS。可以采用Spark Streaming之类的工具来达成实时数据处理和存储。

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

JavaPairRDD<string string=""> 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<string string="">(),
    "org.apache.hadoop.mapred.lib.MultipleTextOutputFormat",
    new Configuration(false)
);
</string></string></code>

登录后复制