如何使用 KStreams 将数据从 Kafka 主题写入文件？

Question

我正在尝试使用 Java 在 Eclipse 中创建一个 KStream 应用程序。现在我指的是互联网上可用于 KStreams 的字数统计程序并对其进行修改。

我想要的是我从输入主题读取的数据应该写入文件而不是写入另一个输出主题。

但是当我尝试将 KStream/KTable 打印到本地文件时，我在输出文件中得到以下条目：

org.apache.kafka.streams.kstream.internals.KStreamImpl@4c203ea1

如何实现将 KStream 的输出重定向到文件？

下面是代码：

package KStreamDemo.kafkatest;

package org.apache.kafka.streams.examples.wordcount;

import org.apache.kafka.clients.consumer.ConsumerConfig;

import org.apache.kafka.common.serialization.Serdes;

import org.apache.kafka.streams.KafkaStreams;

import org.apache.kafka.streams.StreamsBuilder;

import org.apache.kafka.streams.StreamsConfig;

import org.apache.kafka.streams.kstream.KStream;

import org.apache.kafka.streams.kstream.KTable;

import org.apache.kafka.streams.kstream.KeyValueMapper;

import org.apache.kafka.streams.kstream.Produced;

import org.apache.kafka.streams.kstream.ValueMapper;

import java.util.Arrays;

import java.util.Locale;

import java.util.Properties;

import java.util.concurrent.CountDownLatch;

public class TemperatureDemo {

public static void main(String[] args) throws Exception {

    Properties props = new Properties();

    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-wordcount");

    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "34.73.184.104:9092");

    props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0);

    props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());

Answer 1

这是不正确的System.out.println("OUTPUT:"+ counts);您需要执行counts.foreach，然后将消息打印到文件中。将 Kafka 流输入打印到控制台？（只需更新以写入文件）但是，将流写入主题可能更好。并使用 Kafka Connect 写入文件。这是一种更符合行业标准的模式。鼓励 Kafka Streams 仅在 Kafka 内的主题之间移动数据，而不是与外部系统（或文件系统）集成使用您想要的主题信息进行编辑connect-file-sink.properties，然后bin/connect-standalone config/connect-file-sink.properties