SparkStreaming结合Kafka使用

spark自带的example中就有streaming结合kafka使用的案例:

$SPARK_HOME/examples/src/main/scala/org/apache/spark/examples/streaming/KafkaWordCount.scala

使用方法参见代码描述:

Usage: KafkaWordCount <zkQuorum> <group> <topics> <numThreads>
<zkQuorum> is a list of one or more zookeeper servers that make quorum
<group> is the name of kafka consumer group
<topics> is a list of one or more kafka topics to consume from
<numThreads> is the number of threads the kafka consumer should use

Example:
`$ bin/run-example org.apache.spark.examples.streaming.KafkaWordCount zoo01,zoo02,zoo03 my-consumer-group topic1,topic2 1`

运行步骤:

1、启动ZK

zkServer.sh start

2、启动KAFKA SERVER

kafka-server-start.sh  $KAFKA_HOME/config/server.properties &  

3、运行Producer

run-example org.apache.spark.examples.streaming.KafkaWordCountProducer hadoop000:9092 test 3 5

参数描述:

  hadoop000:9092表示producer的地址和端口;

  test表示topic;

  3表示每秒发多少条消息;

  5表示每条消息中有几个单词;

4、运行Consumer

run-example org.apache.spark.examples.streaming.KafkaWordCount hadoop000:2181 test-consumer-group test 1

参数描述:

  hadoop000:2181表示zookeeper的监听地址;

  test-consumer-group表示consumer-group的名称,必须和$KAFKA_HOME/config/consumer.properties中的group.id的配置内容一致;

  test表示topic;

  1表示线程数;

注意观察consumer控制台的数据输出


参考许鹏博客

时间: 2024-10-30 01:10:58

SparkStreaming结合Kafka使用的相关文章

SparkStreaming与Kafka整合遇到的问题及解决方案

前言 最近工作中是做日志分析的平台,采用了sparkstreaming+kafka,采用kafka主要是看中了它对大数据量处理的高性能,处理日志类应用再好不过了,采用了sparkstreaming的流处理框架 主要是考虑到它本身是基于spark核心的,以后的批处理可以一站式服务,并且可以提供准实时服务到elasticsearch中,可以实现准实时定位系统日志. 实现 Spark-Streaming获取kafka数据的两种方式-Receiver与Direct的方式. 一. 基于Receiver方式

sparkStreaming 读kafka的数据

目标:sparkStreaming每2s中读取一次kafka中的数据,进行单词计数. topic:topic1 broker list:192.168.1.126:9092,192.168.1.127:9092,192.168.1.128:9092 1.首先往一个topic中实时生产数据. 代码如下: 代码功能:每秒向topic1发送一条消息,一条消息里包含4个单词,单词之间用空格隔开. 1 package kafkaProducer 2 3 import java.util.HashMap 4

SparkStreaming消费kafka数据

概要:本例子为SparkStreaming消费kafka消息的例子,实现的功能是将数据实时的进行抽取.过滤.转换,然后存储到HDFS中. 实例代码 package com.fwmagic.test import com.alibaba.fastjson.{JSON, JSONException} import org.apache.kafka.common.serialization.StringDeserializer import org.apache.spark.SparkConf imp

SparkStreaming整合kafka的补充

(1)SparkStreaming 整合 kafka 两种方式对比 Direct 方式的优缺点分析 : 优点: 简化并行(Simplified Parallelism).不现需要创建以及 union 多输入源,Kafka topic 的partition 与 RDD 的 partition 一一对应. 高效(Efficiency).基于 Receiver-based 的方式保证数据零丢失(zero-data loss)需要配置 spark.streaming.receiver.writeAhea

spark-streaming读kafka数据到hive遇到的问题

在项目中使用spark-stream读取kafka数据源的数据,然后转成dataframe,再后通过sql方式来进行处理,然后放到hive表中, 遇到问题如下,hive-metastor在没有做高可用的情况下,有时候会出现退出,这个时候,spark streaminG的微批作业就会失败, 然后再启重动hive-metastore进程后,作业继续正常执行,数据就有丢失. 分析如下: 第一步,观察日志发现, 我原来的代码这么写的: xx.foreachRdd(rdd=> processRdd(rdd

第91讲:sparkStreaming基于kafka的Direct详解

1:Direct方式特点: 1)Direct的方式是会直接操作kafka底层的元数据信息,这样如果计算失败了,可以把数据重新读一下,重新处理.即数据一定会被处理.拉数据,是RDD在执行的时候直接去拉数据. 2)由于直接操作的是kafka,kafka就相当于你底层的文件系统.这个时候能保证严格的事务一致性,即一定会被处理,而且只会被处理一次.而Receiver的方式则不能保证,因为Receiver和ZK中的数据可能不同步,spark Streaming可能会重复消费数据,这个调优可以解决,但显然没

Spark实战(一)SparkStreaming集成Kafka

Spark Streaming + Kafka集成指南 Kafka项目在版本0.8和0.10之间引入了一个新的消费者API,因此有两个独立的相应Spark Streaming包可用.请选择正确的包, 请注意,0.8集成与后来的0.9和0.10代理兼容,但0.10集成与早期的代理不兼容. 注意:从Spark 2.3.0开始,不推荐使用Kafka 0.8支持. Spark Streaming从Kafka接收数据,转换为spark streaming中的数据结构Dstream.数据接收方式有两种 :1

第89课:SparkStreaming on Kafka之Kafka解析和安装实战

本篇博文将从以下方面组织内容: 1. Kafka解析 2. 消息组件Kafka 3. Kafka安装 实验搭建所需要的软件: kafka_2.10-0.9.0.1 Zookeeper集群已经安装好.在上一篇博文有安装步骤,不清楚的朋友可以参考下. 一:Kafka解析 1. Kafka是生产者和消费者模式中广播概念,Kafka也可以实现队列的方式. 2. Kafka不仅是一个消息中间键,还是一个存储系统,可以将流进来的数据存储一段时间.这就与传统的流式处理不一样,传统的流式处理处理完数据之后就消失

sparkStreaming读取kafka的两种方式

概述 Spark Streaming 支持多种实时输入源数据的读取,其中包括Kafka.flume.socket流等等.除了Kafka以外的实时输入源,由于我们的业务场景没有涉及,在此将不会讨论.本篇文章主要着眼于我们目前的业务场景,只关注Spark Streaming读取Kafka数据的方式. Spark Streaming 官方提供了两种方式读取Kafka数据: 一是Receiver-based Approach.该种读取模式官方最先支持,并在Spark 1.2提供了数据零丢失(zero-d