kafka复习:(25)kafka stream

这篇具有很好参考价值的文章主要介绍了kafka复习:(25)kafka stream。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

一、java代码:

package com.cisdi.dsp.modules.metaAnalysis.rest.kafka2023;

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.*;

import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;

public class KafkaTest25 {
    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, "k8s-master: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());

        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> source = builder.stream("streams-plaintext-input");

        KTable<String, Long> counts = source.flatMapValues(value -> Arrays.asList(value.toLowerCase().split(" ")))
                .groupBy((key, value) -> value).count();

        counts.toStream().to("streams-wordcount-output", Produced.with(Serdes.String(), Serdes.Long()));

        final KafkaStreams streams = new KafkaStreams(builder.build(), props);
        final CountDownLatch latch = new CountDownLatch(1);
        Runtime.getRuntime().addShutdownHook(new Thread("streams-wordcount-shutdown-hook") {
            public void run() {
                streams.close();
                latch.countDown();
            }
        });

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

        System.exit(0);
    }
}

二、启动console producer:

bin/kafka-console-producer.sh --broker-list xx.xx.xx.xx:9092 --topic streams-plaintext-input

三、启动console consumer:

./kafka-console-consumer.sh --bootstrap-server xx.xx.xx.xx:9092 \
    --topic streams-wordcount-output \
    --from-beginning \
    --formatter kafka.tools.DefaultMessageFormatter \
    --property print.key=true \
    --property print.value=true \
    --property key.deserializer=org.apache.kafka.common.serialization.StringDeserializer \
    --property value.deserializer=org.apache.kafka.common.serialization.LongDeserializer

四、在producer端输入字符串(空格分割),看consumer输出文章来源地址https://www.toymoban.com/news/detail-694565.html

到了这里,关于kafka复习:(25)kafka stream的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • Kafka Stream 流式计算

    1.1 概念 一般流式计算会与批量计算相比较。在流式计算模型中,输入是持续的,可以认为在时间上是无界的,也就意味着,永远拿不到全量数据去做计算。同时,计算结果是持续输出的,也即计算结果在时间上也是无界的。流式计算一般对实时性要求较高,同时一般是先定

    2024年01月23日
    浏览(41)
  • 深入理解Kafka Stream

    作者:禅与计算机程序设计艺术 Apache Kafka是一个开源流处理平台,它提供了一个分布式、高吞吐量、可靠的消息传递系统。Kafka Streams是一个基于Kafka的客户端库,它允许开发人员在Kafka集群中实时地进行计算。本文将通过一个Kafka Stream应用的例子,带领读者对Kafka Stream背后的

    2024年02月06日
    浏览(40)
  • Spark Streaming 整合 Kafka

    本专栏案例代码和数据集链接: https://download.csdn.net/download/shangjg03/88477827 Spark 针对 Kafka 的不同版本,提供了两套整合方案:`spark-streaming-kafka-0-8` 和 `spark-streaming-kafka-0-10`,其主要区别如下: 本文使用的 Kafka 版本为 `kafka_2.12-2.2.0`,故采用第二种方式进行整合。

    2024年02月06日
    浏览(50)
  • Kafka的重要组件,谈谈流处理引擎Kafka Stream

    上手第一关,手把手教你安装kafka与可视化工具kafka-eagle Kafka是什么,以及如何使用SpringBoot对接Kafka 架构必备能力——kafka的选型对比及应用场景 Kafka存取原理与实现分析,打破面试难关 防止消息丢失与消息重复——Kafka可靠性分析及优化实践 我们前面介绍了很多kafka本身的特

    2024年02月05日
    浏览(25)
  • 第八篇——Kafka Streams源码解读

    作者:禅与计算机程序设计艺术 Kafka Streams是一个开源分布式流处理平台,它可以让你轻松处理实时数据流。通过Kafka Streams API可以轻松创建、部署和运行复杂的实时流处理应用程序。虽然Kafka Stream提供了许多高级功能,但其底层原理却十分简单易懂,在学习之余,我们还是需

    2024年02月07日
    浏览(43)
  • Spring Cloud Stream集成Kafka

    Spring Cloud Stream是一个构建消息驱动微服务的框架,抽象了MQ的使用方式, 提供统一的API操作。Spring Cloud Stream通过Binder(绑定器)、inputs/outputs Channel完成应用程序和MQ的解耦。 Binder 负责绑定应用程序和MQ中间件,即指定应用程序是和KafKa交互还是和RabbitMQ交互或者和其他的MQ中间件交

    2024年02月13日
    浏览(49)
  • Structured_Streaming和Kafka整合

    默认情况下,Spark的结构化流支持多种输出方案: File Sink foreach sink 允许对输出的数据进行任意的处理操作,具体如何处理由用户自定义函数决定。对输出的数据一个个进行处理操作。 使用方式主要有二种 方式一: 方式二:这种方式的适用场景是需要和资源打交道的情况(

    2024年01月19日
    浏览(61)
  • Spark Streaming + Kafka构建实时数据流

    1. 使用Apache Kafka构建实时数据流 参考文档链接:https://cloud.tencent.com/developer/article/1814030 2. 数据见UserBehavior.csv 数据解释:本次实战用到的数据集是CSV文件,里面是一百零四万条淘宝用户行为数据,该数据来源是阿里云天池公开数据集 根据这一csv文档运用Kafka模拟实时数据流,

    2024年02月12日
    浏览(44)
  • springboot~kafka-stream实现实时统计

    实时统计,也可以理解为流式计算,一个输入流,一个输出流,源源不断。 Kafka Stream是Apache Kafka从0.10版本引入的一个新Feature。它是提供了对存储于Kafka内的数据进行流式处理和分析的功能。 Kafka Stream提供了一个非常简单而轻量的Library,它可以非常方便地嵌入任意Java应用中

    2024年02月03日
    浏览(31)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包