spark streaming消費kafka數據寫入hdfs避免文件覆蓋方案(java版)

1.寫在前面

在spark streaming+kafka對流式數據處理過程中，往往是spark streaming消費kafka的數據寫入hdfs中，再進行hive映射形成數倉，當然也可以利用sparkSQL直接寫入hive形成數倉。對於寫入hdfs中，如果是普通的rdd則API為saveAsTextFile()，如果是PairRDD則API為saveAsHadoopFile()。當然高版本的spark可能將這兩個合二為一。這兩種API在spark streaming中如果不自定義的話會導致新寫入的hdfs文件覆蓋歷史寫入的hdfs文件，下面來重現這個問題。

2.saveAsTextFile()寫新寫入的hdfs文件覆蓋歷史寫入的hdfs文件測試代碼

package com.surfilter.dp.timer.job;

import kafka.message.MessageAndMetadata;
import kafka.serializer.StringDecoder;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.streaming.Seconds;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka.KafkaUtils;

import java.text.SimpleDateFormat;
import java.util.*;

public class TestStreaming extends BaseParams {
    public static void main(String args[]) {
        String totalParameterString = null;
        if (null != args && args.length > 0) {
            totalParameterString = args[0];
        }
        if (null != totalParameterString && !"".equals(totalParameterString)) {
            ParameterParse parameterParse = new ParameterParse(totalParameterString);
            SparkConf conf = new SparkConf().setAppName(parameterParse.getSpark_app_name());
            setSparkConf(parameterParse, conf);
            JavaSparkContext sparkContext = new JavaSparkContext(conf);
            JavaStreamingContext streamingContext = new JavaStreamingContext(sparkContext, Seconds.apply(Long.parseLong(parameterParse.getSpark_streaming_duration())));

            JavaInputDStream<String> dStream = KafkaUtils.createDirectStream(streamingContext, String.class, String.class,
                    StringDecoder.class, StringDecoder.class, String.class,
                    generatorKafkaParams(parameterParse), generatorTopicOffsets(parameterParse, "test_20200509"),
                    new Function<MessageAndMetadata<String, String>, String>() {
                        private static final long serialVersionUID = 1L;

                        @Override
                        public String call(MessageAndMetadata<String, String> msgAndMd) throws Exception {
                            return msgAndMd.message();
                        }
                    });

            dStream.foreachRDD(new VoidFunction<JavaRDD<String>>() {
                @Override
                public void call(JavaRDD<String> rdd) throws Exception {
                    JavaRDD<String> saveHdfsRdd = rdd.mapPartitions(new FlatMapFunction<Iterator<String>, String>() {
                        @Override
                        public Iterable<String> call(Iterator<String> iterator) throws Exception {
                            List<String> returnList = new ArrayList<>();
                            while (iterator.hasNext()){
                                String message = iterator.next().toString();
                                returnList.add(message);
                            }
                            return returnList;
                        }
                    });

                    String dt = new SimpleDateFormat("yyyyMMdd").format(new Date());
                    String hour = new SimpleDateFormat("HH").format(new Date());
                    String savePath = "hdfs://gawh220:8020/user/hive/warehouse/rzx_standard.db/meijs_test/dt=" + dt + "/hour=" + hour + "/";
                    saveHdfsRdd.saveAsTextFile(savePath);
                }
            });

            streamingContext.start();
            streamingContext.awaitTermination();
            streamingContext.close();
        }
    }
}

在yarn上執行spark streaming觀察，用命令行的方式往test_20200509的topic手動生產一段測試數據，發現spark streaming立馬檢測到並執行完成

之后查看寫入的hdfs文件

發現hdfs文件寫入正常，也是有數據的。之后不再繼續命令行生產數據，當sprak streaming新的一個批次記錄為0的任務開始執行並執行完成

再觀察寫入的hdfs文件，發現文件依然有，但是文件的內容為空，這就證明了第一批有數據的被覆蓋掉了

為什么被覆蓋？
spark streaming是按照特定的配置時間去一批批的拉取kafka的數據，在寫入的時候也是按照分區的狀態寫入hdfs中的，比如下圖

可以看出三個分區寫成三個文件，每一批寫入都是按照這種方式自動生成文件名並寫入文件中，所以會造成最新一批覆蓋之前的一批

3.利用saveAsHadoopFile()自定義輸出文件格式避免覆蓋問題

package com.surfilter.dp.timer.job;

import com.surfilter.dp.timer.parse.BaseParams;
import com.surfilter.dp.timer.parse.ParameterParse;
import kafka.message.MessageAndMetadata;
import kafka.serializer.StringDecoder;
import org.apache.hadoop.mapred.lib.MultipleTextOutputFormat;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.sql.hive.HiveContext;
import org.apache.spark.streaming.Seconds;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka.KafkaUtils;
import scala.Tuple2;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.List;

public class TestStreaming extends BaseParams {
    public static void main(String args[]) {
        String totalParameterString = null;
        if (null != args && args.length > 0) {
            totalParameterString = args[0];
        }
        if (null != totalParameterString && !"".equals(totalParameterString)) {
            ParameterParse parameterParse = new ParameterParse(totalParameterString);
            SparkConf conf = new SparkConf().setAppName(parameterParse.getSpark_app_name());
            setSparkConf(parameterParse, conf);
            JavaSparkContext sparkContext = new JavaSparkContext(conf);
            JavaStreamingContext streamingContext = new JavaStreamingContext(sparkContext, Seconds.apply(Long.parseLong(parameterParse.getSpark_streaming_duration())));

            JavaInputDStream<String> dStream = KafkaUtils.createDirectStream(streamingContext, String.class, String.class,
                    StringDecoder.class, StringDecoder.class, String.class,
                    generatorKafkaParams(parameterParse), generatorTopicOffsets(parameterParse, "test_20200509"),
                    new Function<MessageAndMetadata<String, String>, String>() {
                        private static final long serialVersionUID = 1L;

                        @Override
                        public String call(MessageAndMetadata<String, String> msgAndMd) throws Exception {
                            return msgAndMd.message();
                        }
                    });


            dStream.foreachRDD(new VoidFunction<JavaRDD<String>>() {
                @Override
                public void call(JavaRDD<String> rdd) {
                    JavaPairRDD<String, String> pairRDD = rdd.mapPartitionsToPair(new PairFlatMapFunction<Iterator<String>, String, String>() {
                        @Override
                        public Iterable<Tuple2<String, String>> call(Iterator<String> iterator) {
                            List<Tuple2<String, String>> returnTuple = new ArrayList<>();
                            while (iterator.hasNext()) {
                                String message = iterator.next().toString();
                                returnTuple.add(new Tuple2<>(message, ""));
                            }
                            return returnTuple;
                        }
                    });

                    String dt = new SimpleDateFormat("yyyyMMdd").format(new Date());
                    String hour = new SimpleDateFormat("HH").format(new Date());
                    String savePath = "hdfs://gawh220:8020/user/hive/warehouse/rzx_standard.db/meijs_test/dt=" + dt + "/hour=" + hour + "/";

                    pairRDD.saveAsHadoopFile(savePath, String.class, String.class, RDDMultipleTextOutputFormat.class);
                }
            });

            streamingContext.start();
            streamingContext.awaitTermination();
            streamingContext.close();
        }
    }
}

class RDDMultipleTextOutputFormat extends MultipleTextOutputFormat {
    private static String system_time = System.currentTimeMillis() + "";

    @Override
    protected String generateFileNameForKeyValue(Object key, Object value, String name) {
        name = system_time + "-" + name;
        return super.generateFileNameForKeyValue(key, value, name);
    }
}

用命令行的方式往test_20200509的topic手動生產一段測試數據，發現spark streaming立馬檢測到並執行完成

之后查看寫入的hdfs文件

發現hdfs文件寫入正常，也是有數據的。之后不再繼續命令行生產數據，當sprak streaming新的一個批次記錄為0的任務開始執行並執行完成

再觀察寫入的hdfs文件，發現並沒有產生新的hdfs文件

再命令行的方式往test_20200509的topic手動生產一段測試數據，發現spark streaming立馬檢測到並執行完成

之后查看寫入的hdfs文件，發現新寫入的hdfs文件是追加到之前的文件的方式並且有數據的，如果之前的文件大小超過hdfs設定的大小，則會追加新的文件方式

說明：這種方式不但可以避免覆蓋問題，而且可以避免大量小文件