KStreamsを使用してKafkaトピックからファイルにデータを書き込む方法

ディジェ:

私は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());
    props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
    System.out.println("#1###################################################################################################################################################################################");
    // setting offset reset to earliest so that we can re-run the demo code with the same pre-loaded data
    // Note: To re-run the demo, you need to use the offset reset tool:
    // https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Streams+Application+Reset+Tool
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

    StreamsBuilder builder = new StreamsBuilder();
    System.out.println("#2###################################################################################################################################################################################");
    KStream<String, String> source = builder.stream("iot-temperature");
    System.out.println("#5###################################################################################################################################################################################");
    KTable<String, Long> counts = source
        .flatMapValues(new ValueMapper<String, Iterable<String>>() {
            @Override
            public Iterable<String> apply(String value) {
                return Arrays.asList(value.toLowerCase(Locale.getDefault()).split(" "));
            }
        })
        .groupBy(new KeyValueMapper<String, String, String>() {
            @Override
            public String apply(String key, String value) {
                return value;
            }
        })
        .count();
    System.out.println("#3###################################################################################################################################################################################");
    System.out.println("OUTPUT:"+ counts);
    System.out.println("#4###################################################################################################################################################################################");
    // need to override value serde to Long type
    counts.toStream().to("iot-temperature-max", Produced.with(Serdes.String(), Serdes.Long()));

    final KafkaStreams streams = new KafkaStreams(builder.build(), props);
    final CountDownLatch latch = new CountDownLatch(1);

    // attach shutdown handler to catch control-c
    Runtime.getRuntime().addShutdownHook(new Thread("streams-wordcount-shutdown-hook") {
        @Override
        public void run() {
            streams.close();
            latch.countDown();
        }
    });

    try {
        streams.start();
        latch.await();
    } catch (Throwable e) {
        System.exit(1);
    }
    System.exit(0);
}

}

OneCricketeer:

これは正しくありません

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

この記事はインターネットから収集されたものであり、転載の際にはソースを示してください。

侵害の場合は、連絡してください[email protected]

編集
0

コメントを追加

0

関連記事

分類Dev

kafkaトピックにcsvファイルを書き込む

分類Dev

処理を使用してArduinoからテキストファイルにデータを書き込む

分類Dev

バッチスクリプトを使用してJSONファイルからデータを取得し、そのデータをテキストファイルに書き込む方法は?

分類Dev

JavaScriptを使用してデータをJSONファイルに書き込む方法

分類Dev

jqueryを使用してJSONファイルにデータを書き込む方法

分類Dev

Puppeteerを使用してファイルにデータを書き込む方法は?

分類Dev

JDBCコネクタを使用してKafkaトピックに書き込む前にデータを暗号化する方法はありますか?

分類Dev

シリアルポートからデータを読み取ってファイルに書き込む方法

分類Dev

Pythonを使用してテキストファイルから特定のデータを抽出し、CSVに書き込む方法

分類Dev

フックをコピーしてファイルに書き込む

分類Dev

akkaストリームを使用してデータをチャンクでファイルに書き込む

分類Dev

Pythonの基本-APIからデータをリクエストし、ファイルに書き込む

分類Dev

Nodetmpパッケージを使用してバッファからファイルを書き込む方法

分類Dev

Pythonを使用してファイルに例外トレースバックを書き込む方法

分類Dev

ファイルからデータを取得してファイルに書き込む

分類Dev

Javaを使用してPostgresデータベースから収集したデータをテキストファイルに書き込む

分類Dev

ネットワーク共有センターを介してファイルに書き込む方法

分類Dev

LinuxでPOSIXpthreadを使用してファイルからバッファに整数を書き込む方法は?

分類Dev

hdfsコネクタを介してkafkaから書き込むときにavroファイルのサイズを制限する方法はありますか?

分類Dev

実行ボックスからターミナルを開き、「 -- 」を使用してコマンドを実行し、他のディレクトリにファイルを書き込むことはできますか?

分類Dev

CGI-perlを使用してファイルにデータを書き込む

分類Dev

RandomAccessFileを使用してファイルにデータを書き込む

分類Dev

Pythonを使用してjsonファイルにデータを書き込む

分類Dev

Groovyスクリプトを使用してJenkinからファイルを作成して書き込む方法は?

分類Dev

IDでインクリメントしてPHPを使用してファイルにデータを書き込む方法

分類Dev

Angular4 / 5:typescriptを使用してフォルダ、テキストファイルを作成し、そのファイルにデータを書き込む方法

分類Dev

Pythonを使用してリストから複数のExcelファイルにデータフレームを書き込む

分類Dev

Blobオブジェクトを使用して、大きな生のXMLファイルをOracleデータベースに書き込む方法

分類Dev

xlsxwriterを使用してデータフレームなしでExcelファイルに書き込む方法

Related 関連記事

  1. 1

    kafkaトピックにcsvファイルを書き込む

  2. 2

    処理を使用してArduinoからテキストファイルにデータを書き込む

  3. 3

    バッチスクリプトを使用してJSONファイルからデータを取得し、そのデータをテキストファイルに書き込む方法は?

  4. 4

    JavaScriptを使用してデータをJSONファイルに書き込む方法

  5. 5

    jqueryを使用してJSONファイルにデータを書き込む方法

  6. 6

    Puppeteerを使用してファイルにデータを書き込む方法は?

  7. 7

    JDBCコネクタを使用してKafkaトピックに書き込む前にデータを暗号化する方法はありますか?

  8. 8

    シリアルポートからデータを読み取ってファイルに書き込む方法

  9. 9

    Pythonを使用してテキストファイルから特定のデータを抽出し、CSVに書き込む方法

  10. 10

    フックをコピーしてファイルに書き込む

  11. 11

    akkaストリームを使用してデータをチャンクでファイルに書き込む

  12. 12

    Pythonの基本-APIからデータをリクエストし、ファイルに書き込む

  13. 13

    Nodetmpパッケージを使用してバッファからファイルを書き込む方法

  14. 14

    Pythonを使用してファイルに例外トレースバックを書き込む方法

  15. 15

    ファイルからデータを取得してファイルに書き込む

  16. 16

    Javaを使用してPostgresデータベースから収集したデータをテキストファイルに書き込む

  17. 17

    ネットワーク共有センターを介してファイルに書き込む方法

  18. 18

    LinuxでPOSIXpthreadを使用してファイルからバッファに整数を書き込む方法は?

  19. 19

    hdfsコネクタを介してkafkaから書き込むときにavroファイルのサイズを制限する方法はありますか?

  20. 20

    実行ボックスからターミナルを開き、「 -- 」を使用してコマンドを実行し、他のディレクトリにファイルを書き込むことはできますか?

  21. 21

    CGI-perlを使用してファイルにデータを書き込む

  22. 22

    RandomAccessFileを使用してファイルにデータを書き込む

  23. 23

    Pythonを使用してjsonファイルにデータを書き込む

  24. 24

    Groovyスクリプトを使用してJenkinからファイルを作成して書き込む方法は?

  25. 25

    IDでインクリメントしてPHPを使用してファイルにデータを書き込む方法

  26. 26

    Angular4 / 5:typescriptを使用してフォルダ、テキストファイルを作成し、そのファイルにデータを書き込む方法

  27. 27

    Pythonを使用してリストから複数のExcelファイルにデータフレームを書き込む

  28. 28

    Blobオブジェクトを使用して、大きな生のXMLファイルをOracleデータベースに書き込む方法

  29. 29

    xlsxwriterを使用してデータフレームなしでExcelファイルに書き込む方法

ホットタグ

アーカイブ