1.写在前面
在利用spark计算引擎将kafka或其他源数据组件的数据入hive形成数仓的过程中有两种方式，一种方式是利用spark Rdd的API将数据写入hdfs形成hdfs文件，之后再将文件和hdfs文件和hive表做加载映射。第二种方式是利用sparkSQL将获取的数据Rdd转换成dataFrame，再将dataFrame写成缓存表，最后利用sparkSQL直接插入hive表中。这两种方式各有各自的优点。但大多数开发者更倾向于后者一次编码一步到位的方式。而对于利用sparkSQL写hive表官方有两种常见的API，第一种是利用JavaBean做映射，第二种是利用StructType创建Schema做映射，下面根据代码来分析这两种API 。
2.样例数据
原始数据：
tom,1
jim,2
lily,3
lucy,4
写入hive数据
字段 : word num
值 : tom 1
jim 2
lily 3
lucy 4
3.JavaBean做映射方式
String hiveTableColumns = "word,num";
dStream.foreachRDD(new VoidFunction<JavaRDD<String>>() {
@Override
public void call(JavaRDD<String> rdd) throws Exception {
JavaRDD<TestBean> beanJavaRDD = rdd.mapPartitions(new FlatMapFunction<Iterator<String>, TestBean>() {
@Override
public Iterable<TestBean> call(Iterator<String> iterator) throws Exception {
List<TestBean> beans = new ArrayList<>();
while (iterator.hasNext()){
String message = iterator.next().toString();
TestBean bean = new TestBean();
bean.setWord(message.split(",",-1)[0]);
bean.setNum(message.split(",",-1)[1]);
beans.add(bean);
}
return beans;
}
});
DataFrame dataFrame = session.createDataFrame(beanJavaRDD, TestBean.class);
dataFrame.registerTempTable("temp_test");
session.sql("insert into test partition(create_time_p=" + new SimpleDateFormat("yyyyMMdd").format(new Date())
+ ") select " + hiveTableColumns + " from temp_test");
}
});
public class TestBean implements Serializable {
private String word;
private String num;
public String getWord() {
return word;
}
public void setWord(String word) {
this.word = word;
}
public String getNum() {
return num;
}
public void setNum(String num) {
this.num = num;
}
}
3.利用StructType创建Schema做映射方式
String hiveTableColumns = "word,num";
dStream.foreachRDD(new VoidFunction<JavaRDD<String>>() {
@Override
public void call(JavaRDD<String> rdd) throws Exception {
JavaRDD<Row> rowJavaRDD = rdd.mapPartitions(new FlatMapFunction<Iterator<String>, Row>() {
@Override
public Iterable<Row> call(Iterator<String> iterator) throws Exception {
List<Row> rowList = new ArrayList<>();
while (iterator.hasNext()){
String message = iterator.next().toString();
rowList.add(RowFactory.create(message.split(",", -1)));
}
return rowList;
}
});
DataFrame dataFrame = session.createDataFrame(rowJavaRDD, getSchema(hiveTableColumns.split(",",-1)));
dataFrame.registerTempTable("temp_test");
session.sql("insert into " + databaseAndTableName + " partition(create_time_p=" + new SimpleDateFormat("yyyyMMdd").format(new Date())
+ ") select " + hiveTableColumns + " from temp_test");
}
});
public static StructType getSchema(String[] columns) {
List<StructField> schemaFields = new ArrayList<>();
for (int i = 0; i < columns.length - 1; i++) {
schemaFields.add(DataTypes.createStructField(columns[i], DataTypes.StringType, true));
}
return DataTypes.createStructType(schemaFields);
}
4.对比这两种方式
这两种方式实现方式都相对简单，也比较简洁。对于很多大数据初学者可能首先会想到第一种方式。但是第一种方式不具备通用性，也就是新增一种类型数据。又需要新建bean,然后这里JavaRDD<TestBean> beanJavaRDD需要动态，这里DataFrame dataFrame = session.createDataFrame(beanJavaRDD, TestBean.class);也比较麻烦。最后发现根本无法通用多种类型的数据，如果数据有几百种类，这种方式就不够通用，每一类数据都需要对应的程序。而第二种方式就可以通用了，只需要将数据的字段抽取配置，一个类是可以兼容无论多少种数据的。所以在开发过程中还是推荐第二种方式。但是第一种方式也有自己的优点，不会出现字段与值对应错乱的问题。而第二种方式可能稍不小心会出现字段与值错乱的问题。