SparkStreaming 写入 Hive 遇到的问题记录

需求:SparkStreaming 实时写入 Hive
关于怎么写,网上一大堆,我简单点列下代码:

SparkConf sparkConf = new SparkConf().setAppName("sparkStreaming-order").setMaster(SPARK_MASTER);
sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
                .set("spark.streaming.kafka.maxRatePerPartition", "500")
                .set("spark.kryo.registrator", "com.ykc.bean.input.MyRegistrator") //序列化ConsumerRecord类
                .set("hive.metastore.uris", HIVE_METASTORE_URIS)
                .set("spark.sql.warehouse.dir", HIVE_WAREHOUSE_DIR)
                .set("hive.exec.dynamic.partition", "true")
                .set("hive.exec.max.dynamic.partitions", "2048")
                .set("hive.exec.dynamic.partition.mode", "nonstrict");
SparkSession ss = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate();
JavaStreamingContext jsc = new JavaStreamingContext(new JavaSparkContext(ss.sparkContext()), Durations.seconds(30));
// 注意这里有个问题,orderStream在最外面定义它为static或transient,原因在sparkStreaming使用sql这篇博客中有讲过
// 还有手动提交offset,前面也有提过
orderStream = KafkaUtils.createDirectStream(jsc, LocationStrategies.PreferConsistent(),ConsumerStrategies.Subscribe(Lists.newArrayList(topic), kafkaParams, getOffsets(topic)));
// checkpoint
jsc.checkpoint(SPARK_CHECKPOINT_DIR + "/order");
orderStream.checkpoint(Durations.seconds(SPARK_CHECKPOINT_INTERVAL));
orderStream.foreachRDD(new VoidFunction>>() {

	private static final long serialVersionUID = 1L;

	@Override
    public void call(JavaRDD> javaRDD) throws Exception {
		// 处理数据的逻辑
		// 获取到一个javaRDD,然后用sparksql写入hive
		Dataset orderDataSet = ss.createDataFrame(orderRDD, OrderKafkaMessage.class);
        orderDataSet.createOrReplaceTempView("tmp_order");
        String hiveDatabase = PropConfig.getProperty("ykc.hive.database");
        String sql = "insert into " + hiveDatabase + ".ods_pile_log_order " +
                            "select orderStatus,tradeSeq,gunId,startTime,endTime," +
                            "totalPower,chargeDetails,topPower,peakPower,flatPower," +
                            "valleyPower,totalSeviceMoney,totalElecMoney,sumElectricCharge," +
                            "activityParkedDiscountAmount,recordId,uid,meterValueEnd,activityChargedDiscountAmount," +
                            "flowPlanId,stopReason,sumServerCharge,plateNo,connectorPower,chargeLast,userName," +
                            "equipmentID,chargingSource,meterValueStart,startChargeSeq,stationID,gunCodePartition,createTime," +
                            "updateTime,dt from tmp_order";
     	ss.sql(sql);

	// 还有提交offset的代码,略过
		
	}
}

咋一看这代码没什么问题,也能跑起来,但是在做压测的时候发现,在数据量大的情况下,跑的很慢,设置的处理间隔是 30s,Durations.seconds(30);
一个批次处理 1W+ 数据的情况下,执行的很慢,查看 sparkUI 界面发现,主要的耗时都在 ss.sql(sql) 这行代码上。
然后上网找资料,为什么 sparksql 执行 hive 写入的 sql 这么慢?
有些文章说,ss.sql(insert into table) 在执行过程中有三部,一个是 select,一个是 write,一个是 load。主要是最后一个 load 最慢。然后就考虑可以先写入 hdfs,在用 hive 做同步,因为 spark 直接写 hdfs 是很快的。
第一次解决,修改代码如下:

// 先按照年月日分区,再按照时分秒分区
String ymd = DateUtil.format(new Date(), "yyyyMMdd");
String hms = DateUtil.format(new Date(), "HHmmss");
String finalLocation = hiveTpLocation + "/ymd=" + ymd + "/hms=" + hms;
orderRDD.repartition(1).saveAsTextFile(finalLocation );
ss.sql("use " + hiveDatabase);
ss.sql("ALTER TABLE ods_pile_order ADD PARTITION(ymd='" + ymd + "',hms='" + hms + "') location '" + finalLocation + "'")

一些说明:

  • 为什么设置 repartition(1),是为了确保写入 hdfs 的时候只生成一个文件

  • 为什么分区设置的细,为了确保每次 alter table 的时候不在同一个分区,不然会报错:分区已经存在!

  • saveAsTextFile()这个方法写的是 text 格式的文件,写入文件的格式它会调用实体类的 toString(),所以我们重写了 toString() 方法,把属性的值按照 ‘|’ 来分隔开来

  • hive 创表语句要注意按照 ‘|’ 来分隔并且指定分区,比如:

partition by(ymd string,hms string) row format delimited fields terminated by '|';

结果:确实能在 hive 表里查到数据,而且速度很快,一批次处理 40W+ 的数据只花了 20s,批次间隔为 30s,不会造成后面批次的等待延迟。问题似乎解决了,但是总觉的在 hdfs 中存储 text 格式不是最优的方法,我们一天的数据量在估算了后在 1.5G 到 2G 之间,用不了多久磁盘就不够用了(因为公司穷,在阿里云上不愿意多花钱)。所以最优的应该是存 ORCFILE,它能达到 70% 的压缩比,所以在此修改代码:

// RDD转为Dataset
Dataset dataFrame = ss.createDataFrame(orderRDD, OrderKafkaMessage.class);.repartition(1);
// 写入hdfs,用orc格式
dataFrame.write().format("orc").save(fullPath1);
ss.sql("use " + hiveDatabase);
String sql = "ALTER TABLE ods_pile_order ADD PARTITION(ymd='" + ymd + "',hms='" + hms + "')LOCATION '" + fullPath1 + "'";
ss.sql(sql);

但是有个问题,写入到 hdfs 的是什么格式,刚才说如果是 saveAsTextFile()是按照我们实体类的 toString() 来写。那 ORC 又是什么样的呢,经过百度后得知,是自动按照‘\001’分割的,所以创表语句改了下fields terminated by '\001';
满心欢喜打包上服务器跑,结果发现 hive 表里确实有数据,但是很多都是 NULL,只有少部分有值,这个就很奇怪了,经过一大波百度后找到了点问题:我是用 Dataset 去直接写入 hdfs 的,这个 Dataset 的 schema 应该是跟我们的实体类 OrderKafkaMessage 中的属性名是一致的,可是我 hive 表中的字段跟实体类属性不一样的(比如实体类是 chargingPower,hive 字段是 charging_power),所以我需要自定义一个 schema 给我的 Dataset,schema 中的属性名要和 hive 建表语句中的字段名一致,修改代码如下:

// 定义StructType对象
StructType schema = DataTypes.createStructType(new StructField[]{
		DataTypes.createStructField("order_status", DataTypes.StringType, true),
		DataTypes.createStructField("trade_seq", DataTypes.StringType, true),
		// ...这里很多字段就省略,注意和hive表的字段名一致即可
 });
 // rdd的泛型是OrderKafkaMessage
 JavaRDD rowJavaRDD = rdd.mapPartitions(new FlatMapFunction, Row>() {
    @Override
    public Iterator call(Iterator v1) {
        List list = Lists.newArrayList();
        while (v1.hasNext()) {
            OrderKafkaMessage next = v1.next();
            Object[] array = new Object[0];
            try {
                array = CommonUtil.bean2Array(next);
            } catch (IllegalAccessException e) {
                log.error("反射转换出错:{}", next);
            }
            Row row = RowFactory.create(array);
            list.add(row);
        }
        return list.iterator();
    }
});
// schema顺序和传入的 数组值顺序一致
// 比如schema 中第一个是order_status,第二个是trade_seq,上面的array数组中第一个就是OrderKafkaMessage实体类中的orderStatus,因为是反射获取的,所以两边顺序一致
Dataset dataFrame = ss.createDataFrame(rowJavaRDD, schema);

// 因为在创建Row对象的时候要传入一个个的值,不想写很多重复代码,所以自定义了一个方法用反射获取对象的值组成一个数组传进去,公用方法如下:
@Slf4j
public class CommonUtil {

    public static Object[] bean2Array(Object object) throws IllegalAccessException {
        Field[] fields = object.getClass().getDeclaredFields();
        Object[] array = new Object[fields.length];
        for (int i = 0; i < fields.length; i++) {
            fields[i].setAccessible(true);
            array[i] = fields[i].get(object);
        }
        return array;
    }
}
// 然后就是跟上面的一样,把Dataset以ORC格式写入hdfs中。。。

这里还遇到个关于反射的小插曲:
因为我的实体类要序列化,所以implements Serializable,添加了序列 keyprivate static final long serialVersionUID = 1L;
然后再反射调用的时候把这个 serialVersionUID 作为类的一个属性获取到了,结果一直报类型转换错误:Long 不能转为 String。我也不知道为啥会这样,然后就去掉这行代码就可以了。至于去掉这个序列化的 key 会导致一些问题,这里还没有解决,因为也没遇到过。阅读文章的大佬可以给些建议。
还有遇到的问题:
写入成功后,然后用 spark 读取 hive 中的这个订单数据,始终读取不到,一直报错,大致意思就是读取不到表中 Column 的值。最后发现问题所在,OrderKafkaMessage 中有些属性是 double 类型,但是建 hive 表的时候字段都是 String 类型,所以读取不到。所以 hive 建立外表的时候字段类型要和 hdfs 中数值的类型一样,double 就是 double,int 就是 int。

至此,基本上代码就是这样了,上了生产后效率上也还行,hive 中也顺利查到数据。
后续会一直监控作业运行情况,如果遇到其他生产问题也会及时记录补充…