原创

Kafka Streams以及数据清洗小案例

温馨提示:
本文最后更新于 2019年03月21日,已超过 2,087 天没有更新。若文章内的图片失效(无法正常加载),请留言反馈或直接联系我

Kafka Streams 概述

Kafka Streams。Apache Kafka开源项目的一个组成部分。是一个功能强大,易于使用的库。用于在Kafka上构建高可分布式、拓展性,容错的应用程序。

Kafka Streams 特点

1)功能强大

高扩展性,弹性,容错

2)轻量级

无需专门的集群

一个库,而不是框架

3)完全集成

100%的Kafka 0.10.0版本兼容

易于集成到现有的应用程序

4)实时性

毫秒级延迟

并非微批处理

窗口允许乱序数据

允许迟到数据

Kafka Streams 数据清洗案例

实时处理字符带有”>>>”前缀的内容。例如输入”111>>>222”,最终处理成“222”

package com.lzhpo.kafka.kafkaStreams;

import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;

/**
 * <p> Author:lzhpo </p>
 * <p> Title:</p>
 * <p> Description:
 * 具体业务处理
 * </p>
 */
public class LogProcessor implements Processor<byte[], byte[]> {
    private ProcessorContext context;

    @Override
    public void init(ProcessorContext context) {
        this.context = context;
    }

    @Override
    public void process(byte[] key, byte[] value) {
        String input = new String(value);

        // 如果包含“>>>”则只保留该标记后面的内容
        if (input.contains(">>>")) {
            input = input.split(">>>")[1].trim();
            // 输出到下一个topic
            context.forward("logProcessor".getBytes(), input.getBytes());
        }else{
            context.forward("logProcessor".getBytes(), input.getBytes());
        }
    }

    @Override
    public void punctuate(long timestamp) {

    }

    @Override
    public void close() {

    }
}
package com.lzhpo.kafka.kafkaStreams;

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorSupplier;
import org.apache.kafka.streams.processor.TopologyBuilder;

import java.util.Properties;

/**
 * <p> Author:lzhpo </p>
 * <p> Title:【Kafka数据清洗案例】</p>
 * <p> Description:
 * 
 * 实时处理字符带有”>>>”前缀的内容。例如输入”111>>>222”,最终处理成“222”
 *
 * 生产者(定义输入的topic):
 * kafka-console-producer.sh --broker-list localhost:9092 --topic first
 *
 * 消费者(定义输出的topic):
 * kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic second
 * </p>
 */
public class APP {

    public static void main(String[] args) {

        // 定义输入的topic
        String from = "first";
        // 定义输出的topic
        String to = "second";

        // 设置参数
        Properties settings = new Properties();
        settings.put(StreamsConfig.APPLICATION_ID_CONFIG, "logFilter");
        settings.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.200.111:9092");

        StreamsConfig config = new StreamsConfig(settings);

        // 构建拓扑
        TopologyBuilder builder = new TopologyBuilder();

        builder.addSource("SOURCE", from)
                .addProcessor("PROCESS", new ProcessorSupplier<byte[], byte[]>() {

                    @Override
                    public Processor<byte[], byte[]> get() {
                        // 具体分析处理
                        return new LogProcessor();
                    }
                }, "SOURCE")
                .addSink("SINK", to, "PROCESS");

        // 创建kafka stream
        KafkaStreams streams = new KafkaStreams(builder, config);
        streams.start();
    }

}

让APP一直保持运行:

开启一个生产者(定义输入的topic):

kafka-console-producer.sh --broker-list localhost:9092 --topic first

发送消息

开启一个消费者(定义输出的topic):

kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic second

Consumer接收到的消息:

可以看到:

生产者发送111>>>222,消费者收到222;

生产者发送666>>>999,消费者收到999;

生产者发送222>>>666,消费者收到666;

生产者发送其它不相关的,消费者收到原型。

本文目录