spark-kafka direct方式读取和receiver方式读取的区别

心已赠人 2022-01-06 04:47 223阅读 0赞

> ## 区别 ##

Spark-Streaming获取kafka数据的两种方式-**Receiver与Direct的方式**，**可以从代码中简单理解成Receiver方式是通过zookeeper来连接kafka队列，Direct方式是直接连接到kafka的节点上获取数据了**。

### 一、基于Receiver的方式 ###

这种方式使用Receiver来获取数据。Receiver是使用Kafka的高层次Consumer API来实现的。receiver从Kafka中获取的数据都是存储在Spark Executor的内存中的，然后Spark Streaming启动的job会去处理那些数据。

然而，在默认的配置下，这种方式可能会因为底层的失败而丢失数据。如果要启用高可靠机制，让数据零丢失，就必须启用Spark Streaming的预写日志机制（Write Ahead Log，WAL）。该机制会同步地将接收到的Kafka数据写入分布式文件系统（比如HDFS）上的预写日志中。所以，即使底层节点出现了失败，也可以使用预写日志中的数据进行恢复。  
需要注意的要点

**1、Kafka中的topic的partition，与Spark中的RDD的partition是没有关系的。所以，在KafkaUtils.createStream()中，提高partition的数量，只会增加一个Receiver中，读取partition的线程的数量。不会增加Spark处理数据的并行度**。

**2、可以创建多个Kafka输入DStream，使用不同的consumer group和topic，来通过多个receiver并行接收数据。**

**3、如果基于容错的文件系统，比如HDFS，启用了预写日志机制，接收到的数据都会被复制一份到预写日志中。因此，在KafkaUtils.createStream()中，设置的持久化级别是StorageLevel.MEMORY\_AND\_DISK\_SER。**

### 二、基于Direct的方式 ###

这种新的不基于Receiver的直接方式，是在Spark 1.3中引入的，从而能够确保更加健壮的机制。替代掉使用Receiver来接收数据后，这种方式会周期性地查询Kafka，来获得每个topic+partition的最新的offset，从而定义每个batch的offset的范围。当处理数据的job启动时，就会使用Kafka的简单consumer api来获取Kafka指定offset范围的数据。

这种方式有如下优点：

**1、简化并行读取：**

**如果要读取多个partition，不需要创建多个输入DStream然后对它们进行union操作**。**Spark会创建跟Kafka partition一样多的RDD partition，并且会并行从Kafka中读取数据。所以在Kafka partition和RDD partition之间，有一个一对一的映射关系。**

**2、高性能**：

如果要保证零数据丢失，在基于receiver的方式中，需要开启WAL机制。这种方式其实效率低下，因为数据实际上被复制了两份，Kafka自己本身就有高可靠的机制，会对数据复制一份，而这里又会复制一份到WAL中。而基于direct的方式，不依赖Receiver，不需要开启WAL机制，只要Kafka中作了数据的复制，那么就可以通过Kafka的副本进行恢复。

**3、一次且仅一次的事务机制**：

基于receiver的方式，是使用Kafka的高阶API来在ZooKeeper中保存消费过的offset的。这是消费Kafka数据的传统方式。这种方式配合着WAL机制可以保证数据零丢失的高可靠性，**但是却无法保证数据被处理一次且仅一次**，可能会处理两次。因为Spark和ZooKeeper之间可能是不同步的。

**4、降低资源。**  
Direct不需要Receivers，其申请的Executors全部参与到计算任务中；而Receiver-based则需要专门的Receivers来读取Kafka数据且不参与计算。因此相同的资源申请，Direct 能够支持更大的业务。

**5、降低内存**。  
Receiver-based的Receiver与其他Exectuor是异步的，并持续不断接收数据，对于小业务量的场景还好，如果遇到大业务量时，需要提高Receiver的内存，但是参与计算的Executor并无需那么多的内存。而Direct 因为没有Receiver，而是在计算时读取数据，然后直接计算，所以对内存的要求很低。实际应用中我们可以把原先的10G降至现在的2-4G左右。

**6、鲁棒性更好。**  
Receiver-based方法需要Receivers来异步持续不断的读取数据，因此遇到网络、存储负载等因素，导致实时任务出现堆积，但Receivers却还在持续读取数据，此种情况很容易导致计算崩溃。Direct 则没有这种顾虑，其Driver在触发batch 计算任务时，才会读取数据并计算。队列出现堆积并不会引起程序的失败。

基于direct的方式，使用kafka的简单api，Spark Streaming自己就负责追踪消费的offset，并保存在checkpoint中。Spark自己一定是同步的，因此可以保证数据是消费一次且仅消费一次。

> ## **代码** ##

### **1、spark接收kafka数据 - receiver模式 - java代码** ###

/*
     * Licensed to the Apache Software Foundation (ASF) under one or more
     * contributor license agreements.  See the NOTICE file distributed with
     * this work for additional information regarding copyright ownership.
     * The ASF licenses this file to You under the Apache License, Version 2.0
     * (the "License"); you may not use this file except in compliance with
     * the License.  You may obtain a copy of the License at
     *
     *    http://www.apache.org/licenses/LICENSE-2.0
     *
     * Unless required by applicable law or agreed to in writing, software
     * distributed under the License is distributed on an "AS IS" BASIS,
     * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
     * See the License for the specific language governing permissions and
     * limitations under the License.
     */
    
    package com.wzq;
    
    import java.util.Arrays;
    import java.util.Iterator;
    import java.util.Map;
    import java.util.HashMap;
    import java.util.regex.Pattern;
    
    import scala.Tuple2;
    
    import org.apache.spark.SparkConf;
    import org.apache.spark.api.java.function.FlatMapFunction;
    import org.apache.spark.api.java.function.Function;
    import org.apache.spark.api.java.function.Function2;
    import org.apache.spark.api.java.function.PairFunction;
    import org.apache.spark.streaming.Duration;
    import org.apache.spark.streaming.api.java.JavaDStream;
    import org.apache.spark.streaming.api.java.JavaPairDStream;
    import org.apache.spark.streaming.api.java.JavaPairReceiverInputDStream;
    import org.apache.spark.streaming.api.java.JavaStreamingContext;
    import org.apache.spark.streaming.kafka.KafkaUtils;
    
    /**
     * Consumes messages from one or more topics in Kafka and does wordcount.
     *
     * Usage: JavaKafkaWordCount <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
     *
     * To run this example:
     *   `$ bin/run-example org.apache.spark.examples.streaming.JavaKafkaWordCount zoo01,zoo02, \
     *    zoo03 my-consumer-group topic1,topic2 1`
     */
    
    public final class JavaKafkaWordCount {
      private static final Pattern SPACE = Pattern.compile(" ");
    
      private JavaKafkaWordCount() {
      }
    
      public static void main(String[] args) throws Exception {
        if (args.length < 4) {
          System.err.println("Usage: JavaKafkaWordCount <zkQuorum> <group> <topics> <numThreads>");
          System.exit(1);
        }
    // 设置local模式，【2】表示启动两个线程
    //    SparkConf sparkConf = new SparkConf().setAppName("JavaKafkaWordCount").setMaster("local[2]");
        SparkConf sparkConf = new SparkConf().setAppName("JavaKafkaWordCount");
        // Create the context with 2 seconds batch size
        JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, new Duration(2000));
    
        int numThreads = Integer.parseInt(args[3]);
        Map<String, Integer> topicMap = new HashMap<>();
        String[] topics = args[2].split(",");
        for (String topic: topics) {
          topicMap.put(topic, numThreads);
        }
    
        JavaPairReceiverInputDStream<String, String> messages =
                KafkaUtils.createStream(jssc, args[0], args[1], topicMap);
    
        JavaDStream<String> lines = messages.map(new Function<Tuple2<String, String>, String>() {
          @Override
          public String call(Tuple2<String, String> tuple2) {
            return tuple2._2();
          }
        });
    
        JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
        	 @Override
             public Iterator<String> call(String line) {
                 return Arrays.asList(line.split(" ")).iterator();
             }
        });
    
        JavaPairDStream<String, Integer> wordCounts = words.mapToPair(
          new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String s) {
              return new Tuple2<>(s, 1);
            }
          }).reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer i1, Integer i2) {
              return i1 + i2;
            }
          });
        wordCounts.print();
        jssc.start();
        jssc.awaitTermination();
      }
    }

### **2、spark接收kafka数据 - direct模式 - java代码** ###

package com.wzq;
    
    /*
     * Licensed to the Apache Software Foundation (ASF) under one or more
     * contributor license agreements.  See the NOTICE file distributed with
     * this work for additional information regarding copyright ownership.
     * The ASF licenses this file to You under the Apache License, Version 2.0
     * (the "License"); you may not use this file except in compliance with
     * the License.  You may obtain a copy of the License at
     *
     *    http://www.apache.org/licenses/LICENSE-2.0
     *
     * Unless required by applicable law or agreed to in writing, software
     * distributed under the License is distributed on an "AS IS" BASIS,
     * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
     * See the License for the specific language governing permissions and
     * limitations under the License.
     */
    
    import java.util.HashMap;
    import java.util.HashSet;
    import java.util.Arrays;
    import java.util.Iterator;
    import java.util.Map;
    import java.util.Set;
    import java.util.regex.Pattern;
    
    import scala.Tuple2;
    
    import kafka.serializer.StringDecoder;
    
    import org.apache.spark.SparkConf;
    import org.apache.spark.api.java.function.*;
    import org.apache.spark.streaming.api.java.*;
    import org.apache.spark.streaming.kafka.KafkaUtils;
    import org.apache.spark.streaming.Durations;
    
    /**
     * Consumes messages from one or more topics in Kafka and does wordcount. Usage:
     * JavaDirectKafkaWordCount <brokers> <topics> <brokers> is a list of one or
     * more Kafka brokers <topics> is a list of one or more kafka topics to consume
     * from
     *
     * Example: $ bin/run-example streaming.JavaDirectKafkaWordCount
     * broker1-host:port,broker2-host:port \ topic1,topic2
     */
    
    public final class JavaDirectKafkaWordCount {
    	private static final Pattern SPACE = Pattern.compile(" ");
    
    	public static void main(String[] args) throws Exception {
    //		args = new String[]{"kafkasit02broker01.cn.com:9092,kafkasit02broker02.cn.com:9092,kafkasit02broker03.cn.com:9092","ssmp_data_sit"};
    		if (args.length < 2) {
    			System.err.println("Usage: JavaDirectKafkaWordCount <brokers> <topics>\n"
    					+ "  <brokers> is a list of one or more Kafka brokers\n"
    					+ "  <topics> is a list of one or more kafka topics to consume from\n\n");
    			System.exit(1);
    		}
    
    		String brokers = args[0];
    		String topics = args[1];
    
    		// Create context with a 2 seconds batch interval
    		SparkConf sparkConf = new SparkConf().setAppName("JavaDirectKafkaWordCount");
    		JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, Durations.seconds(2));
    
    		Set<String> topicsSet = new HashSet<>(Arrays.asList(topics.split(",")));
    		Map<String, String> kafkaParams = new HashMap<>();
    		kafkaParams.put("metadata.broker.list", brokers);
    
    		// Create direct kafka stream with brokers and topics
    		JavaPairInputDStream<String, String> messages = KafkaUtils.createDirectStream(jssc, String.class, String.class,
    				StringDecoder.class, StringDecoder.class, kafkaParams, topicsSet);
    
    		// Get the lines, split them into words, count the words and print
    		JavaDStream<String> lines = messages.map(new Function<Tuple2<String, String>, String>() {
    			@Override
    			public String call(Tuple2<String, String> tuple2) {
    				return tuple2._2();
    			}
    		});
    		JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
    			@Override
    			public Iterator<String> call(String x) {
    				return Arrays.asList(SPACE.split(x)).iterator();
    			}
    		});
    		JavaPairDStream<String, Integer> wordCounts = words.mapToPair(new PairFunction<String, String, Integer>() {
    			@Override
    			public Tuple2<String, Integer> call(String s) {
    				return new Tuple2<>(s, 1);
    			}
    		}).reduceByKey(new Function2<Integer, Integer, Integer>() {
    			@Override
    			public Integer call(Integer i1, Integer i2) {
    				return i1 + i2;
    			}
    		});
    		wordCounts.print();
    
    		// Start the computation
    		jssc.start();
    		jssc.awaitTermination();
    	}
    }

**3、pom文件**

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    	<modelVersion>4.0.0</modelVersion>
    	<groupId>myselfProject1</groupId>
    	<artifactId>myselfProject1</artifactId>
    	<version>0.0.1-SNAPSHOT</version>
    	<name>myselfProject1</name>
    	<description>myselfProject1</description>
    	<properties>
    		<org.springframework-version>4.0.6.RELEASE</org.springframework-version>
    		<common-version>2.6.0</common-version>
    		<zookeeper-version>3.4.0</zookeeper-version>
    	</properties>
    	<dependencies>
    		
    		<dependency>
    			<groupId>org.apache.spark</groupId>
    			<artifactId>spark-streaming_2.11</artifactId>
    			<version>2.1.0</version>
    		</dependency>
    
    		<dependency>
    			<groupId>org.apache.spark</groupId>
    			<artifactId>spark-core_2.10</artifactId>
    			<version>2.1.0</version>
    		</dependency>
    
    		<dependency>
    			<groupId>org.apache.hadoop</groupId>
    			<artifactId>hadoop-client</artifactId>
    			<version>2.2.0</version>
    		</dependency>
    		<dependency>
    			<groupId>org.apache.spark</groupId>
    			<artifactId>spark-streaming_2.10</artifactId>
    			<version>2.1.0</version>
    		</dependency>
    		<dependency>
    			<groupId>org.apache.spark</groupId>
    			<artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
    			<version>2.1.0</version>
    		</dependency>
    
    	</dependencies>
    	
    	
    	<build>
    		<plugins>
    			<plugin>
    				<groupId>org.apache.maven.plugins</groupId>
    				<artifactId>maven-compiler-plugin</artifactId>
    				<configuration>
    					<source>1.7</source>
    					<target>1.7</target>
    					<encoding>UTF-8</encoding>
    				</configuration>
    			</plugin>
    
    			
    			<plugin>
    				<artifactId>maven-assembly-plugin</artifactId>
    				<configuration>
    					<appendAssemblyId>false</appendAssemblyId>
    					<descriptorRefs>
    						<descriptorRef>jar-with-dependencies</descriptorRef>
    					</descriptorRefs>
    					<archive>
    						<manifest>
    							
    							<mainClass>com.wzq.JavaDirectKafkaWordCount</mainClass>
    						</manifest>
    					</archive>
    				</configuration>
    				<executions>
    					<execution>
    						<id>make-assembly</id>
    						<phase>package</phase>
    						<goals>
    							<goal>assembly</goal>
    						</goals>
    					</execution>
    				</executions>
    			</plugin>
    		</plugins>
    	</build>
    </project>

转：[https://blog.csdn.net/wzqllwy/article/details/78869889][https_blog.csdn.net_wzqllwy_article_details_78869889]

[https_blog.csdn.net_wzqllwy_article_details_78869889]: https://blog.csdn.net/wzqllwy/article/details/78869889