Spark从入门到入土(四):SparkStreaming集成kafka

简介: Spark从入门到入土(四):SparkStreaming集成kafka

一、SparkStreaming概念


SparkStreaming是一个准实时的数据处理框架,支持对实时数据流进行可扩展、高吞吐量、容错的流处理,SparkStreaming可以从kafka、HDFS等中获取数据,经过SparkStreaming数据处理后保存到HDFS、数据库等。


image.png

sparkStreaming


spark streaming接收实时输入数据流,并将数据分为多个微批,然后由spark engine进行处理,批量生成最终结果流。


image.png

处理流程


二、基本操作


2.1初始化StreamingContext


Durations指定接收数据的延迟时间,多久触发一次job

SparkConf conf = new SparkConf().setMaster("local").setAppName("alarmCount");
JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(10));


2.2基本操作


1:streamingcontext.start() 开始接受数据


2:streamingContext.stop() 停止


2.3注意的点


1:上下文启动后,不能重新设置或添加新的流式计算


2:一个JVM进程中只能有一个StreamingContext 存活


2.4DStream


DStream是离散数据流,是由一系列RDD组成的序列


1:每个InputDStream对应一个接收器(文件流不需要接收器),一个接收器也只接受一个单一的数据流,但是SparkStreaming应用中可以创建多个输入流


2:每个接收器占用一个核,应用程序的核数要大于接收器数量,如果小于数据将无法全部梳理


三、从kafka中读取数据


通过KafkaUtils从kafka读取数据,读取数据有两种方式,createDstream和createDirectStream。


3.1:createDstream:基于Receiver的方式


1: kafka数据持续被运行在Spark workers/executors 中的Kafka Receiver接受,这种方式使用的是kafka的高阶用户API


2:接受到的数据存储在Spark workers/executors内存以及WAL(Write Ahead Logs), 在数据持久化到日志后,kafka接收器才会更新zookeeper中的offset


3:接受到的数据信息及WAL位置信息被可靠存储,失败时用于重新读取数据。


image.png

createDstream读取数流程


3.2:createDirectStream 直接读取方式


这种方式下需要自行管理offset,可以通过checkpoint或者数据库方式管理


image.png

1.png


SparkStreaming


public class SparkStreaming {
    private static String CHECKPOINT_DIR = "/Users/dbq/Documents/checkpoint";
    public static void main(String[] args) throws InterruptedException {
        //初始化StreamingContext
        SparkConf conf = new SparkConf().setMaster("local").setAppName("alarmCount");
        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(10));
        jssc.checkpoint(CHECKPOINT_DIR);
        Map<String, Object> kafkaParams = new HashMap<>();
        kafkaParams.put("metadata.broker.list", "172.*.*.6:9092,172.*.*.7:9092,172.*.*.8:9092");
        kafkaParams.put("bootstrap.servers", "172.*.*.6:9092,172.*.*.7:9092,172.*.*.8:9092");
        kafkaParams.put("key.deserializer", StringDeserializer.class);
        kafkaParams.put("value.deserializer", StringDeserializer.class);
        kafkaParams.put("group.id", "alarmGroup");
        kafkaParams.put("auto.offset.reset", "latest");
        kafkaParams.put("enable.auto.commit", true);
        Collection<String> topics = Arrays.asList("alarmTopic");
        JavaInputDStream<ConsumerRecord<String, String>> messages =
                KafkaUtils.createDirectStream(
                        jssc,
                        LocationStrategies.PreferConsistent(),
                        ConsumerStrategies.<String, String>Subscribe(topics, kafkaParams)
                );
        JavaDStream<String> lines = messages.map((Function<ConsumerRecord<String, String>, String>) record -> record.value());
        lines.foreachRDD((VoidFunction<JavaRDD<String>>) record -> {
            List<String> list = record.collect();
            for (int i = 0; i < list.size(); i++) {
                writeToFile(list.get(i));
            }
        });
        lines.print();
        jssc.start();
        jssc.awaitTermination();
        System.out.println("----------------end");
    }
    //将结果写入到文件,也可以写入到MongoDB或者HDFS等
    private synchronized static void writeToFile(String content) {
        String fileName = "/Users/dbq/Documents/result.txt";
        FileWriter writer = null;
        try {
            writer = new FileWriter(fileName, true);
            writer.write(content + " \r\n");
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                if (writer != null) {
                    writer.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}


Kafka的集成


生产者配置类

public class KafkaProducerConfig {
    @Value("${spring.kafka.bootstrap-servers}")
    private String broker;
    @Value("${spring.kafka.producer.acks}")
    private String acks;
    @Value("${spring.kafka.producer.retries}")
    private Integer retries;
    @Value("${spring.kafka.producer.batch-size}")
    private Integer batchSize;
    @Value("${spring.kafka.producer.buffer-memory}")
    private long bufferMemory;
    public Map<String, Object> getConfig() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, broker);
        props.put(ProducerConfig.ACKS_CONFIG, acks);
        props.put(ProducerConfig.RETRIES_CONFIG, retries);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return props;
    }
}


Kafka生产者

@Component
public class Producer {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    public void send(Message message) {
        kafkaTemplate.send("alarmTopic", JSONObject.toJSONString(message));
    }
}


配置kafkaTemplate

@Component
public class PushMessageConfig {
   @Autowired
   private PushProducerListener producerListener;
   @Autowired
   private KafkaProducerConfig producerConfig;
   @Bean
   public KafkaTemplate<String, String> kafkaTemplate() {
       @SuppressWarnings({ "unchecked", "rawtypes" })
       ProducerFactory<String, String> factory = new DefaultKafkaProducerFactory<>(producerConfig.getConfig());
       KafkaTemplate<String, String> kafkaTemplate = new KafkaTemplate<>(factory, true);
       kafkaTemplate.setProducerListener(producerListener);
       kafkaTemplate.setDefaultTopic("alarmTopic");
       return kafkaTemplate;
   }
}


配置生产者监听

@Component
public class PushProducerListener implements ProducerListener<String, String> {
    private Logger logger = LoggerFactory.getLogger(PushProducerListener.class);
    @Override
    public void onSuccess(String topic, Integer partition, String key, String value,
                          RecordMetadata recordMetadata) {
        // 数据成功发送到消息队列
        System.out.println("发送成功:" + value);
        logger.info("onSuccess. " + key + " : " + value);
    }
    @Override
    public void onError(String topic, Integer partition, String key, String value,
                        Exception exception) {
        logger.error("onError. " + key + " : " + value);
        logger.error("catching an error when sending data to mq.", exception);
        // 发送到消息队列失败,直接在本地处理
    }
    @Override
    public boolean isInterestedInSuccess() {
        // 发送成功后回调onSuccess,false则不回调
        return true;
    }
}


相关文章
|
1月前
|
分布式计算 大数据 Apache
ClickHouse与大数据生态集成:Spark & Flink 实战
【10月更文挑战第26天】在当今这个数据爆炸的时代,能够高效地处理和分析海量数据成为了企业和组织提升竞争力的关键。作为一款高性能的列式数据库系统,ClickHouse 在大数据分析领域展现出了卓越的能力。然而,为了充分利用ClickHouse的优势,将其与现有的大数据处理框架(如Apache Spark和Apache Flink)进行集成变得尤为重要。本文将从我个人的角度出发,探讨如何通过这些技术的结合,实现对大规模数据的实时处理和分析。
108 2
ClickHouse与大数据生态集成:Spark & Flink 实战
|
2月前
|
消息中间件 分布式计算 NoSQL
大数据-104 Spark Streaming Kafka Offset Scala实现Redis管理Offset并更新
大数据-104 Spark Streaming Kafka Offset Scala实现Redis管理Offset并更新
44 0
|
2月前
|
消息中间件 存储 分布式计算
大数据-103 Spark Streaming Kafka Offset管理详解 Scala自定义Offset
大数据-103 Spark Streaming Kafka Offset管理详解 Scala自定义Offset
98 0
|
12天前
|
消息中间件 Java Kafka
什么是Apache Kafka?如何将其与Spring Boot集成?
什么是Apache Kafka?如何将其与Spring Boot集成?
43 5
|
14天前
|
消息中间件 Java Kafka
Spring Boot 与 Apache Kafka 集成详解:构建高效消息驱动应用
Spring Boot 与 Apache Kafka 集成详解:构建高效消息驱动应用
22 1
|
1月前
|
分布式计算 大数据 OLAP
AnalyticDB与大数据生态集成:Spark & Flink
【10月更文挑战第25天】在大数据时代,实时数据处理和分析变得越来越重要。AnalyticDB(ADB)是阿里云推出的一款完全托管的实时数据仓库服务,支持PB级数据的实时分析。为了充分发挥AnalyticDB的潜力,将其与大数据处理工具如Apache Spark和Apache Flink集成是非常必要的。本文将从我个人的角度出发,分享如何将AnalyticDB与Spark和Flink集成,构建端到端的大数据处理流水线,实现数据的实时分析和处理。
62 1
|
2月前
|
消息中间件 分布式计算 Kafka
大数据-102 Spark Streaming Kafka ReceiveApproach DirectApproach 附带Producer、DStream代码案例
大数据-102 Spark Streaming Kafka ReceiveApproach DirectApproach 附带Producer、DStream代码案例
58 0
|
4月前
|
消息中间件 安全 Java
Spring Boot 基于 SCRAM 认证集成 Kafka 的详解
【8月更文挑战第4天】本文详解Spring Boot结合SCRAM认证集成Kafka的过程。SCRAM为Kafka提供安全身份验证。首先确认Kafka服务已启用SCRAM,并准备认证凭据。接着,在`pom.xml`添加`spring-kafka`依赖,并在`application.properties`中配置Kafka属性,包括SASL_SSL协议与SCRAM-SHA-256机制。创建生产者与消费者类以实现消息的发送与接收功能。最后,通过实际消息传递测试集成效果与认证机制的有效性。
167 4
|
4月前
|
消息中间件 Kafka 数据处理
实时数据流处理:Dask Streams 与 Apache Kafka 集成
【8月更文第29天】在现代数据处理领域,实时数据流处理已经成为不可或缺的一部分。随着物联网设备、社交媒体和其他实时数据源的普及,处理这些高吞吐量的数据流成为了一项挑战。Apache Kafka 作为一种高吞吐量的消息队列服务,被广泛应用于实时数据流处理场景中。Dask Streams 是 Dask 库的一个子模块,它为 Python 开发者提供了一个易于使用的实时数据流处理框架。本文将介绍如何将 Dask Streams 与 Apache Kafka 结合使用,以实现高效的数据流处理。
87 0
|
2月前
|
消息中间件 存储 运维
为什么说Kafka还不是完美的实时数据通道
【10月更文挑战第19天】Kafka 虽然作为数据通道被广泛应用,但在实时性、数据一致性、性能及管理方面存在局限。数据延迟受消息堆积和分区再平衡影响;数据一致性难以达到恰好一次;性能瓶颈在于网络和磁盘I/O;管理复杂性涉及集群配置与版本升级。