Kafka数据清洗ETL

2023-11-21 08:33:46

Kafka Streams

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

2.Kafka Streams特点

　　1)功能强大:高扩展性,弹性,容错

　　2)轻量级:无需专门的集群,一个库,而不是框架.

　　3)完全集成:100%的Kafka 0.10版本兼容;易于集成到现有的程序

　　4)实时性:毫秒级延迟,并非微批处理,窗口允许乱序数据,允许迟到数据

3.

当前已经有非常多的流式处理系统，最知名且应用最多的开源流式处理系统有Spark Streaming和Apache Storm。Apache Storm发展多年，应用广泛，提供记录级别的处理能力，当前也支持SQL on Stream。而Spark Streaming基于Apache Spark，可以非常方便与图计算，SQL处理等集成，功能强大，对于熟悉其它Spark应用开发的用户而言使用门槛低。另外，目前主流的Hadoop发行版，如Cloudera和Hortonworks，都集成了Apache Storm和Apache Spark，使得部署更容易。

既然Apache Spark与Apache Storm拥用如此多的优势，那为何还需要Kafka Stream呢？主要有如下原因。

第一，Spark和Storm都是流式处理框架，而Kafka Stream提供的是一个基于Kafka的流式处理类库。框架要求开发者按照特定的方式去开发逻辑部分，供框架调用。开发者很难了解框架的具体运行方式，从而使得调试成本高，并且使用受限。而Kafka Stream作为流式处理类库，直接提供具体的类给开发者调用，整个应用的运行方式主要由开发者控制，方便使用和调试。

Kafka数据清洗ETL

第二，虽然Cloudera与Hortonworks方便了Storm和Spark的部署，但是这些框架的部署仍然相对复杂。而Kafka Stream作为类库，可以非常方便的嵌入应用程序中，它对应用的打包和部署基本没有任何要求。

第三，就流式处理系统而言，基本都支持Kafka作为数据源。例如Storm具有专门的kafka-spout，而Spark也提供专门的spark-streaming-kafka模块。事实上，Kafka基本上是主流的流式处理系统的标准数据源。换言之，大部分流式系统中都已部署了Kafka，此时使用Kafka Stream的成本非常低。

第四，使用Storm或Spark Streaming时，需要为框架本身的进程预留资源，如Storm的supervisor和Spark on YARN的node manager。即使对于应用实例而言，框架本身也会占用部分资源，如Spark Streaming需要为shuffle和storage预留内存。但是Kafka作为类库不占用系统资源。

第五，由于Kafka本身提供数据持久化，因此Kafka Stream提供滚动部署和滚动升级以及重新计算的能力。

第六，由于Kafka Consumer Rebalance机制，Kafka Stream可以在线动态调整并行度。

 

案例:数据清洗

需求描述:

实时处理单词带有”>>>”前缀的内容。例如输入”lxz>>>lexue”，最终处理成“lexue”;

0) pom文件导入

 <dependencies>

        <!-- https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-common -->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>2.7.3</version>

        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-hdfs -->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>2.7.3</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-client -->
        <!-- Winodws下提交至Yarn上运行，改客户端是2.6.1s
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>2.6.1</version>
        </dependency>
-->

        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.10</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.mrunit/mrunit MRUnit测试 -->
        <dependency>
            <groupId>org.apache.mrunit</groupId>
            <artifactId>mrunit</artifactId>
            <version>0.9.0-incubating</version>
            <classifier>hadoop2</classifier>
            <scope>test</scope>
        </dependency>

        <dependency>
            <groupId>com.sun</groupId>
            <artifactId>tools</artifactId>
            <version>1.8.0</version>
            <scope>system</scope>
            <systemPath>${env.JAVA_HOME}/lib/tools.jar</systemPath>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.hbase/hbase -->
        <dependency>
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase</artifactId>
            <version>1.3.2</version>
            <type>pom</type>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.hbase/hbase-common -->
        <dependency>
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase-common</artifactId>
            <version>1.3.2</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.hbase/hbase-server -->
        <dependency>
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase-server</artifactId>
            <version>1.3.2</version>
        </dependency>


        <!-- https://mvnrepository.com/artifact/org.apache.hbase/hbase-client -->
        <dependency>
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase-client</artifactId>
            <version>1.3.2</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.zookeeper/zookeeper -->
        <dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>3.4.6</version>
            <type>pom</type>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.glassfish.jersey.core/jersey-client -->
        <dependency>
            <groupId>org.glassfish.jersey.core</groupId>
            <artifactId>jersey-client</artifactId>
            <version>2.26</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>0.10.0.0</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-streams -->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-streams</artifactId>
            <version>0.10.0.0</version>
        </dependency>

    </dependencies>

1) 创建主程序

package com.lxz.kafka;

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;

public class Application {
    public static void main(String[] args) {
        //1.定义输入的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,"hadoop1:9092");

        StreamsConfig config = new StreamsConfig(settings);

        //扩建拓扑
        TopologyBuilder builder = new TopologyBuilder();
        builder.addSource("SOURCE",from)
                .addProcessor("PROCESS",new ProcessorSupplier<byte[],byte[]>(){
                    public Processor<byte[],byte[]> get(){
                        //具体分析
//                        return new LogProcessor();
                        return new LogProcessor();
                    }
                },"SOURCE")
                .addSink("SINK",to,"PROCESS");

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

2) 具体业务处理

package com.lxz.kafka;

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

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 l) {

    }

    @Override
    public void close() {

    }
}

3) 报错信息处理

如果遇到log4j提示报警,则是因为缺少了log4j的配置文件,在resources中创建log4j.properties并写入

log4j.rootLogger=INFO,console,dailyFile
# TODO 发布到阿里云记得添加，另外控制台不输出(只输出warn或者error信息)

# log4j.logger.org.mybatis = INFO
log4j.logger.com.imooc.mapper=INFO

log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.encoding=UTF-8
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%-d{yyyy-MM-dd HH:mm:ss,SSS} [%t] [%l] - [%p] %m%n

# 定期滚动日志文件，每天都会生成日志
log4j.appender.dailyFile=org.apache.log4j.DailyRollingFileAppender
log4j.appender.dailyFile.encoding=UTF-8
log4j.appender.dailyFile.Threshold=INFO
# TODO 本地日志地址，正式环境请务必切换为阿里云地址
log4j.appender.dailyFile.File=C:/logs/maven-ssm-alipay/log.log4j
log4j.appender.dailyFile.DatePattern=‘.‘yyyy-MM-dd
log4j.appender.dailyFile.layout=org.apache.log4j.PatternLayout
log4j.appender.dailyFile.layout.ConversionPattern=%-d{yyyy-MM-dd HH:mm:ss,SSS} [%t] [%l] - [%p] %m%n

4) 测试

　　1.运行Idea主程序

　　2.启动hadoop1 2 3,三台服务器,zk,kafka集群

　　3.在hadoop2上启动生产者: bin/kafka-console-producer.sh --broker-list hadoop1:9092 --topic first

　　4.在hadoop3上启动消费者: bin/kafka-console-consumer.sh --zookeeper hadoop1:2181 --from-beginning --topic second

　　5.在hadoop2上生产者下输入:lxz>>>lexue

　　6.查看hadoop3上的消费者是否成功消费到了:lexue

 

 生产者端:

Kafka数据清洗ETL

 

 消费者端:

Kafka数据清洗ETL

 

Kafka数据清洗ETL

上一篇：ModalResult是指一?模式窗体（form.showmodal）的返回值


下一篇：百度地图API 快速上手实例 ！！！