Kafka Stream 流式计算

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

1 实时流式计算

1.1 概念

一般流式计算会与批量计算相比较。在流式计算模型中,输入是持续的,可以认为在时间上是无界的,也就意味着,永远拿不到全量数据去做计算。同时,计算结果是持续输出的,也即计算结果在时间上也是无界的。流式计算一般对实时性要求较高,同时一般是先定义目标计算,然后数据到来之后将计算逻辑应用于数据。同时为了提高计算效率,往往尽可能采用增量计算代替全量计算。

Kafka Stream 流式计算,java,linq,c#

流式计算就相当于上图的右侧扶梯,是可以源源不断的产生数据,源源不断的接收数据,没有边界。

1.2 应用场景

  • 日志分析

    网站的用户访问日志进行实时的分析,计算访问量,用户画像,留存率等等,实时的进行数据分析,帮助企业进行决策

  • 大屏看板统计

    可以实时的查看网站注册数量,订单数量,购买数量,金额等。

  • 公交实时数据

    可以随时更新公交车方位,计算多久到达站牌等

  • 实时文章分值计算

    头条类文章的分值计算,通过用户的行为实时文章的分值,分值越高就越被推荐。

1.3 技术方案选型

  • Hadoop

    Kafka Stream 流式计算,java,linq,c#
  • Apche Storm

    Storm 是一个分布式实时大数据处理系统,可以帮助我们方便地处理海量数据,具有高可靠、高容错、高扩展的特点。是流式框架,有很高的数据吞吐能力。

  • Kafka Stream

    可以轻松地将其嵌入任何Java应用程序中,并与用户为其流应用程序所拥有的任何现有打包,部署和操作工具集成。

2 Kafka Stream

2.1 概述

Kafka Stream是Apache Kafka从0.10版本引入的一个新Feature。它是提供了对存储于Kafka内的数据进行流式处理和分析的功能。

Kafka Stream的特点如下:

  • Kafka Stream提供了一个非常简单而轻量的Library,它可以非常方便地嵌入任意Java应用中,也可以任意方式打包和部署

  • 除了Kafka外,无任何外部依赖

  • 充分利用Kafka分区机制实现水平扩展和顺序性保证

  • 通过可容错的state store实现高效的状态操作(如windowed join和aggregation)

  • 支持正好一次处理语义

  • 提供记录级的处理能力,从而实现毫秒级的低延迟

  • 支持基于事件时间的窗口操作,并且可处理晚到的数据(late arrival of records)

  • 同时提供底层的处理原语Processor(类似于Storm的spout和bolt),以及高层抽象的DSL(类似于Spark的map/group/reduce)

Kafka Stream 流式计算,java,linq,c#

2.2 Kafka Streams的关键概念

  • 源处理器(Source Processor):源处理器是一个没有任何上游处理器的特殊类型的流处理器。它从一个或多个kafka主题生成输入流。通过消费这些主题的消息并将它们转发到下游处理器。

  • Sink处理器:sink处理器是一个没有下游流处理器的特殊类型的流处理器。它接收上游流处理器的消息发送到一个指定的Kafka主题

Kafka Stream 流式计算,java,linq,c#

2.3 KStream

(1)数据结构类似于map,如下图,key-value键值对

Kafka Stream 流式计算,java,linq,c#

(2)KStream

Kafka Stream 流式计算,java,linq,c#

KStream数据流(data stream),即是一段顺序的,可以无限长,不断更新的数据集。 数据流中比较常记录的是事件,这些事件可以是一次鼠标点击(click),一次交易,或是传感器记录的位置数据。

KStream负责抽象的,就是数据流。与Kafka自身topic中的数据一样,类似日志,每一次操作都是向其中插入(insert)新数据。

为了说明这一点,让我们想象一下以下两个数据记录正在发送到流中:

(“ alice”,1)->(“” alice“,3)

如果您的流处理应用是要总结每个用户的价值,它将返回4alice。为什么?因为第二条数据记录将不被视为先前记录的更新。(insert)新数据

2.4 Kafka Stream入门案例编写

(1)需求分析,求单词个数(word count)

Kafka Stream 流式计算,java,linq,c#

(2)引入依赖

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams</artifactId>
    <exclusions>
        <exclusion>
            <artifactId>connect-json</artifactId>
            <groupId>org.apache.kafka</groupId>
        </exclusion>
        <exclusion>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
        </exclusion>
    </exclusions>
</dependency>

(3)创建原生的kafka staream入门案例

​
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
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.TimeWindows;
import org.apache.kafka.streams.kstream.ValueMapper;
​
import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;
​
/**
 * 流式处理
 */
public class KafkaStreamQuickStart {
​
    public static void main(String[] args) {
​
        //kafka的配置信心
        Properties prop = new Properties();
        prop.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.130:9092");
        prop.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        prop.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        prop.put(StreamsConfig.APPLICATION_ID_CONFIG,"streams-quickstart");
​
        //stream 构建器
        StreamsBuilder streamsBuilder = new StreamsBuilder();
​
        //流式计算
        streamProcessor(streamsBuilder);
​
​
        //创建kafkaStream对象
        KafkaStreams kafkaStreams = new KafkaStreams(streamsBuilder.build(),prop);
        //开启流式计算
        kafkaStreams.start();
    }
​
    /**
     * 流式计算
     * 消息的内容:hello kafka  hello itcast
     * @param streamsBuilder
     */
    private static void streamProcessor(StreamsBuilder streamsBuilder) {
        //创建kstream对象,同时指定从那个topic中接收消息
        KStream<String, String> stream = streamsBuilder.stream("itcast-topic-input");
        /**
         * 处理消息的value
         */
        stream.flatMapValues(new ValueMapper<String, Iterable<String>>() {
            @Override
            public Iterable<String> apply(String value) {
                return Arrays.asList(value.split(" "));
            }
        })
                //按照value进行聚合处理
                .groupBy((key,value)->value)
                //时间窗口
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                //统计单词的个数
                .count()
                //转换为kStream
                .toStream()
                .map((key,value)->{
                    System.out.println("key:"+key+",vlaue:"+value);
                    return new KeyValue<>(key.key().toString(),value.toString());
                })
                //发送消息
                .to("itcast-topic-out");
    }
}

(4)测试准备

  • 使用生产者在topic为:itcast_topic_input中发送多条消息

  • 使用消费者接收topic为:itcast_topic_out

结果:

  • 通过流式计算,会把生产者的多条消息汇总成一条发送到消费者中输出

2.5 SpringBoot集成Kafka Stream

(1)自定配置参数

​
import lombok.Getter;
import lombok.Setter;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.Topology;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafkaStreams;
import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration;
import org.springframework.kafka.config.KafkaStreamsConfiguration;
​
import java.util.HashMap;
import java.util.Map;
​
/**
 * 通过重新注册KafkaStreamsConfiguration对象,设置自定配置参数
 */
​
@Setter
@Getter
@Configuration
@EnableKafkaStreams
@ConfigurationProperties(prefix="kafka")
public class KafkaStreamConfig {
    private static final int MAX_MESSAGE_SIZE = 16* 1024 * 1024;
    private String hosts;
    private String group;
    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
    public KafkaStreamsConfiguration defaultKafkaStreamsConfig() {
        Map<String, Object> props = new HashMap<>();
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, hosts);
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, this.getGroup()+"_stream_aid");
        props.put(StreamsConfig.CLIENT_ID_CONFIG, this.getGroup()+"_stream_cid");
        props.put(StreamsConfig.RETRIES_CONFIG, 10);
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        return new KafkaStreamsConfiguration(props);
    }
}

修改application.yml文件,在最下方添加自定义配置

kafka:
  hosts: 192.168.200.130:9092
  group: ${spring.application.name}

(2)新增配置类,创建KStream对象,进行聚合

​
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.ValueMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
​
import java.time.Duration;
import java.util.Arrays;
​
@Configuration
@Slf4j
public class KafkaStreamHelloListener {
​
    @Bean
    public KStream<String,String> kStream(StreamsBuilder streamsBuilder){
        //创建kstream对象,同时指定从那个topic中接收消息
        KStream<String, String> stream = streamsBuilder.stream("itcast-topic-input");
        stream.flatMapValues(new ValueMapper<String, Iterable<String>>() {
            @Override
            public Iterable<String> apply(String value) {
                return Arrays.asList(value.split(" "));
            }
        })
                //根据value进行聚合分组
                .groupBy((key,value)->value)
                //聚合计算时间间隔
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                //求单词的个数
                .count()
                .toStream()
                //处理后的结果转换为string字符串
                .map((key,value)->{
                    System.out.println("key:"+key+",value:"+value);
                    return new KeyValue<>(key.key().toString(),value.toString());
                })
                //发送消息
                .to("itcast-topic-out");
        return stream;
    }
}

测试:

启动微服务,正常发送消息,可以正常接收到消息文章来源地址https://www.toymoban.com/news/detail-819388.html

到了这里,关于Kafka Stream 流式计算的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

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

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

相关文章

  • Java 8:Stream API 流式操作(学习)

    Java 8 中的 Stream API 是一组用于对集合数据进行处理的新特性;提供一种以声明式风格对集合进行操作的方式,简化集合的处理,使得代码更加简洁、优雅,并且能够更高效地处理数据; 这种风格将要处理的元素集合看作一种流, 流在管道中传输, 并且可以在管道的节点上进

    2024年02月10日
    浏览(43)
  • 跟我学java|Stream流式编程——并行流

    并行流是 Java 8 Stream API 中的一个特性。它可以将一个流的操作在多个线程上并行执行,以提高处理大量数据时的性能。 在传统的顺序流中,所有的操作都是在单个线程上按照顺序执行的。而并行流则会将流的元素分成多个小块,并在多个线程上并行处理这些小块,最后将结

    2024年01月25日
    浏览(64)
  • 让代码变得优雅简洁的神器:Java8 Stream流式编程

    本文主要基于实际项目常用的Stream Api流式处理总结。 因笔者主要从事风控反欺诈相关工作,故而此文使用比较熟悉的三要素之一的 【手机号】黑名单 作代码案例说明。 我在项目当中,很早就开始使用Java 8的流特性进行开发了,但是一直都没有针对这块进行开发总结。这次

    2023年04月10日
    浏览(42)
  • 【Java基础教程】(三十)Java新特性篇 · 第十讲: Stream流——释放流式编程的效率与优雅,狂肝万字只为透彻讲清 Stream流!~

    Java的Stream流是在Java 8中引入的一种用于处理集合数据的功能强大且易于使用的工具,旨在简化集合框架的操作。它的设计目的是为了提供一种更简洁、更灵活和更可读的方式来处理集合数据。 在之前,我们通常使用迭代器或循环来遍历和操作集合元素,这种方式容易出错且

    2024年02月16日
    浏览(38)
  • 流式计算中的多线程处理:如何使用Kafka实现高效的实时数据处理

    作者:禅与计算机程序设计艺术 Apache Kafka 是 Apache Software Foundation 下的一个开源项目,是一个分布式的、高吞吐量的、可扩展的消息系统。它最初由 LinkedIn 开发并于 2011 年发布。与其他一些类似产品相比,Kafka 有着更强大的功能和活跃的社区支持。因此,越来越多的人开始使

    2024年02月12日
    浏览(67)
  • Stream流式编程,让代码变优雅

    一、引言 流式编程的概念和作用 Java 流(Stream)是一连串的元素序列,可以进行各种操作以实现数据的转换和处理。流式编程的概念基于函数式编程的思想,旨在简化代码,提高可读性和可维护性。 Java Stream 的主要作用有以下几个方面: 简化集合操作:使用传统的 for 循环或迭

    2024年02月05日
    浏览(42)
  • ChatGPT流式传输(stream=True)的实现-OpenAI API 流式传输

    默认情况下,当请求OpenAI的API时,整个响应将在生成后一次性发送回来。如果需要的响应比较复杂,就会需要很长时间来等待响应。 为了更快地获得响应,可以在请求API时选择“流式传输”。 要使用流式传输,调用API时设置 stream=True 。这将返回一个对象,以 data-only server-

    2024年02月08日
    浏览(72)
  • axios / fetch 实现 stream 流式请求

    axios 是一个支持node端和浏览器端的易用、简洁且高效的http库。本文主要介绍 axios 如何实现 stream 流式请求,注意这里需要区分 node 环境和浏览器环境。 代码演示: 在浏览器端,axios 是使用 XMLHttpRequest 对象来实现请求,设置 responseType: \\\'stream\\\' 后会出现以下警告⚠️: The pr

    2024年02月12日
    浏览(45)
  • 前端流式(stream)请求,获取持续响应的方式

    在网络请求中,流(stream)表示持续接收数据的过程。使用流,可以逐步获取服务器响应,而不是等待整个响应完成后一次性获取。 Axios 通过监听响应对象的 data 事件,可以实现流式请求,主要的写法有直接 pipe 流和手动监听事件两种。使用流可以实现大文件或数据的流传输,

    2024年03月25日
    浏览(67)
  • Egg.js构建一个stream流式接口服务

    经常需要用到 stream 流式接口服务,比如:大文件下载、日志实时输出等等。本文将介绍如何使用Egg.js构建一个 stream 流式接口服务。 目录结构: index.js 控制器 test.txt 测试文件,最好是20M以上的文件,这样才能看出流式返回的效果 test.sh 测试脚本,用于实时输出日志的测试脚

    2024年02月12日
    浏览(36)

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

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

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

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

二维码1

领取红包

二维码2

领红包