hadoop - Spark 流 : Micro batches Parallel Execution

Question

我们正在从 Kafka 的 spark streaming 中接收数据。一旦在 Spark Streaming 中开始执行，它只执行一个批处理，其余批处理开始在 Kafka 中排队。

Our data is independent and can be processes in Parallel.

我们尝试了具有多个执行器、内核、背压和其他配置的多种配置，但到目前为止没有任何效果。有很多消息在排队，一次只处理了一个微批处理，其余的都留在队列中。

我们希望最大程度地实现并行性，以便没有任何微批处理排队，因为我们有足够的可用资源。那么我们如何通过最大限度地利用资源来减少时间。

// Start reading messages from Kafka and get DStream
final JavaInputDStream<ConsumerRecord<String, byte[]>> consumerStream = KafkaUtils.createDirectStream(
        getJavaStreamingContext(), LocationStrategies.PreferConsistent(),
        ConsumerStrategies.<String, byte[]>Subscribe("TOPIC_NAME",
                sparkServiceConf.getKafkaConsumeParams()));

ThreadContext.put(Constants.CommonLiterals.LOGGER_UID_VAR, CommonUtils.loggerUniqueId());

JavaDStream<byte[]> messagesStream = consumerStream.map(new Function<ConsumerRecord<String, byte[]>, byte[]>() {
    private static final long serialVersionUID = 1L;
    @Override
    public byte[] call(ConsumerRecord<String, byte[]> kafkaRecord) throws Exception {
        return kafkaRecord.value();
    }
});

    // Decode each binary message and generate JSON array
        JavaDStream<String> decodedStream = messagesStream.map(new Function<byte[], String>() {
            private static final long serialVersionUID = 1L;

            @Override
            public String call(byte[] asn1Data) throws Exception {
                if(asn1Data.length > 0) {
                    try (InputStream inputStream = new ByteArrayInputStream(asn1Data);
                            Writer writer = new StringWriter(); ) {


                        ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(asn1Data);
                        GZIPInputStream gzipInputStream = new GZIPInputStream(byteArrayInputStream);

                        byte[] buffer = new byte[1024];
                        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();

                        int len;
                        while((len = gzipInputStream.read(buffer)) != -1) {
                            byteArrayOutputStream.write(buffer, 0, len);
                        }


                        return new String(byteArrayOutputStream.toByteArray());


                    } catch (Exception e) {
//                      
                        producer.flush();

                        throw e;
                    }
                } 

                return null;
            }
        });




// publish generated json gzip to kafka 
        cache.foreachRDD(new VoidFunction<JavaRDD<String>>() {
            private static final long serialVersionUID = 1L;

            @Override
            public void call(JavaRDD<String> jsonRdd4DF) throws Exception {
                //Dataset<Row> json = sparkSession.read().json(jsonRdd4DF);
                if(!jsonRdd4DF.isEmpty()) {
                    //JavaRDD<String> jsonRddDF = getJavaSparkContext().parallelize(jsonRdd4DF.collect());
                    Dataset<Row> json = sparkSession.read().json(jsonRdd4DF);   

                    SparkAIRMainJsonProcessor airMainJsonProcessor = new SparkAIRMainJsonProcessor();
                    airMainJsonProcessor.processAIRData(json, sparkSession);
                }

            }               
        });

        getJavaStreamingContext().start();
        getJavaStreamingContext().awaitTermination();
        getJavaStreamingContext().stop();

我们使用的技术:

HDFS  2.7.1.2.5 
YARN + MapReduce2  2.7.1.2.5 
ZooKeeper  3.4.6.2.5 
Ambari Infra  0.1.0 
Ambari Metrics  0.1.0 
Kafka  0.10.0.2.5 
Knox  0.9.0.2.5 
Ranger  0.6.0.2.5 
Ranger KMS  0.6.0.2.5 
SmartSense  1.3.0.0-1
Spark2  2.0.x.2.5 

我们从差异实验中得到的统计数据:

Experiment 1

num_executors=6
executor_memory=8g
executor_cores=12

100 个文件处理时间 48 分钟

Experiment 2

spark.default.parallelism=12
num_executors=6
executor_memory=8g
executor_cores=12

100 个文件处理时间 8 分钟

Experiment 3

spark.default.parallelism=12
num_executors=6
executor_memory=8g
executor_cores=12

100 个文件处理时间 7 分钟

Experiment 4

spark.default.parallelism=16
num_executors=6
executor_memory=8g
executor_cores=12

100 个文件处理时间 10 分钟

请指教，我们如何才能处理最多，以免排队。

Answer 1

我遇到了同样的问题，我尝试了一些方法来解决这个问题，并得出以下结论:

首先。直觉告诉我们每个执行者必须处理一个批处理，但相反，一次只处理一个批处理，但作业和任务是并行处理的。

可以通过使用 spark.streaming.concurrentjobs 实现多批处理，但它没有记录，仍然需要一些修复。问题之一是保存 Kafka 偏移量。假设我们将此参数设置为 4，并且并行处理 4 个批处理，如果第 3 个批处理在第 4 个批处理之前完成，将会提交哪些 Kafka 偏移量。如果批处理是独立的，则此参数非常有用。

spark.default.parallelism 因为它的名字有时被认为是让事物并行。但它真正的好处在于分布式洗牌操作。尝试不同的数字并为此找到最佳数字。您将在处理时间上获得相当大的差异。这取决于您工作中的洗牌操作。设置得太高会降低性能。从你的实验结果也很明显。

另一种选择是使用 foreachPartitionAsync 代替 RDD 上的 foreach。但我认为 foreachPartition 更好，因为 foreachPartitionAsync 会将作业排队，而批处理似乎已处理，但它们的作业仍在队列中或正在处理中。可能是我没有正确使用它。但它在我的 3 项服务中表现相同。

FAIR spark.scheduler.mode 必须用于具有大量任务的作业，因为将任务循环分配给作业，让较小的任务有机会在处理较大的任务时开始接收资源。

尝试调整您的批处理持续时间+输入大小，并始终将其保持在处理持续时间以下，否则您会看到大量的批处理积压。

这些是我的发现和建议，但是，进行流式传输的配置和方法太多了，通常一套操作对其他操作不起作用。 Spark Streaming 就是学习，将您的经验和预期结合起来以获得一组最佳配置。

希望对您有所帮助。如果有人能具体告诉我们如何合法地并行处理批处理，那将是一个很大的安慰。