Spark2-的序列化JavaSerializerKryoSerializer

9次阅读

共计 1961 个字符，预计需要花费 5 分钟才能阅读完成。

JDK   1.8.0  
Scala 2.11.8  
Spark 2.1.2

官方文档: Data Serialization
spark 默认的序列化器是 JavaSerializer，能够支持所有对象自动的序列化，但是效率比较低。
KryoSerializer 比 JavaSerializer 效率高很多，但是不支持所有对象的序列化（比如？？），在使用时需要手动注册自定义类。如果不注册，性能比 JavaSerializer 更糟糕。
可以配置 spark.kryo.registrationRequired=true 来检测自定义类是否注册

相关配置项

spark.serializer=org.apache.spark.serializer.KryoSerializer
spark.kryo.registrationRequired=true

Java 版 WordCount 示例

import dw.common.util.HdfsHelper;
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.sql.SparkSession;
import scala.Tuple2;
import java.util.Arrays;

public class WordCount {public static void main(String[] args) throws ClassNotFoundException {
        // 输入文件
        String wordFile = "/user/qhy/input/wordcount/idea.txt";
        SparkConf sparkConf = new SparkConf();
        sparkConf.registerKryoClasses(new Class<?>[]{
                java.lang.Class.class,
                Object[].class,
                Class.forName("scala.reflect.ClassTag$$anon$1")
        });
        SparkSession spark = SparkSession.builder()
                .appName("WordCount")
                .config(sparkConf)
                .config("spark.executor.instances", 10)
                .config("spark.executor.memory", "4g")
                .config("spark.executor.cores", 1)
                .config("spark.hadoop.mapreduce.output.fileoutputformat.compress", false)
                .getOrCreate();
        JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());
        JavaRDD<String> hdfstext = jsc.textFile(wordFile);
        // 切分
        JavaRDD<String> words = hdfstext.flatMap(line ->
                                        Arrays.asList(line.split("\\s+")).iterator());
        // 单次计 1
        JavaPairRDD<String, Integer> pairs = words.mapToPair(word -> new Tuple2<>(word, 1));
        // 累加 1
        JavaPairRDD<String, Integer> wordCounts = pairs.reduceByKey((v1, v2) -> v1 + v2);
        // 交换 k, v
        JavaPairRDD<Integer, String> swapWordCounts = wordCounts.mapToPair(tuple2 -> tuple2.swap());
        // 降序
        swapWordCounts = swapWordCounts.sortByKey(false, 1).repartition(1);
        swapWordCounts.map(tuple -> tuple._1 + tuple._2);
        // 保存结果到 HDFS
        String outDir = "/user/qhy/output/wordcount";
        HdfsHelper.deleteDir(jsc, outDir);
        swapWordCounts.saveAsTextFile(outDir);
        jsc.close();}
}