SparkSql实现Mysql到hive的数据流动

68 阅读 0 评论 45 点赞

我是靠谱客的博主包容龙猫，这篇文章主要介绍SparkSql实现Mysql到hive的数据流动，现在分享给大家，希望可以做个参考。

今天去面试了一波，因为调度系统采用了SparkSql实现数据从Mysql到hive，在这一点上面试官很明显很不满我对于Spark的理解，19年的第一个面试就这么挂了。
有问题不怕，怕的是知道了问题还得过且过。现在就来梳理下我的项目是怎么使用Spark导数的
第一步：把mysql中的表放入内存

复制代码

1
2
3
4
5
6
7
8
9

properties.put("user", dbUser);
properties.put("password", dbPassword);
properties.put("driver", dbDriver);
Dataset<Row> bizdateDS = sparkSession.read().jdbc(
dbUrl,
dbTableName,
properties
);

其中：org.apache.spark.sql.Dataset（这里面试官问我怎么把mysql的数据转化到Spark，我没答上来）

第二步：创建数据库与表
2.1 创建库

复制代码

1
2
3
4
5
6
7

String createDBSQL = "CREATE DATABASE IF NOT EXISTS " + hiveDBName + " LOCATION '" + dbPath + "'";
sparkSession.sql(createDBSQL);
```
2.2创建表
分成两步，第一步读取Mysql元数据字段，第二步把这些字段创建出来
2.2.1 读取mysql字段

复制代码

1
2
3
4
5
6
7
8
9
 StructType structType = bizdateDS.schema();
StructField[] structFields = structType.fields();
/*
structField是StructType中的字段。
param：name此字段的名称。
param：dataType此字段的数据类型。
param：nullable指示此字段的值是否为空值。
param：metadata此字段的元数据。 如果未修改列的内容（例如，在选择中），则应在转换期间保留元数据。
*/

复制代码

1
2
3
4

2.2.2
创建字段

复制代码

String sourceType; //Name of the type used in JSON serialization.
String columnName;
String targetType;
StructField structField;
SparkDataTypeEnum sparkDataType;
StringBuilder createBuilder = new StringBuilder(capacity);
createBuilder.append("CREATE TABLE IF NOT EXISTS ").append(realHiveTableName).append(" (");
List<String> dbTableColumns = Lists.newArrayList();
Map<String, String> dbTableColumnTypeMap = Maps.newHashMap();
//把Mysql中的每个字段都提取出来
for (int i = 0, len = structFields.length; i < len; i++) {
structField = structFields[i];
sourceType = structField.dataType().typeName();
columnName = structField.name();
if (sourceType.contains("(")) { //处理类似varchar(20)
sourceType = sourceType.substring(0, sourceType.indexOf("("));
}
sparkDataType = SparkDataTypeEnum.getItemByType(sourceType);
if (null != sparkDataType) {
targetType = sparkDataType.getHiveDataType().getType();
//时间戳字段强转成string字段
if(targetType.equals("timestamps")) targetType.equals("string");
} else {
targetType = HiveDataTypeEnum.STRING.getType();
}
dbTableColumns.add(columnName);
dbTableColumnTypeMap.put(columnName, targetType);
if (i != 0) {
createBuilder.append(",");
}
createBuilder.append(columnName).append(" ").append(targetType);
}
createBuilder.append(") PARTITIONED by (").append(partitionColumn)
.append(" STRING) ");
sparkSession.sql(createTableSQL);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

String sourceType; //Name of the type used in JSON serialization.
String columnName;
String targetType;
StructField structField;
SparkDataTypeEnum sparkDataType;
StringBuilder createBuilder = new StringBuilder(capacity);
createBuilder.append("CREATE TABLE IF NOT EXISTS ").append(realHiveTableName).append(" (");
List<String> dbTableColumns = Lists.newArrayList();
Map<String, String> dbTableColumnTypeMap = Maps.newHashMap();
//把Mysql中的每个字段都提取出来
for (int i = 0, len = structFields.length; i < len; i++) {
structField = structFields[i];
sourceType = structField.dataType().typeName();
columnName = structField.name();
if (sourceType.contains("(")) { //处理类似varchar(20)
sourceType = sourceType.substring(0, sourceType.indexOf("("));
}
sparkDataType = SparkDataTypeEnum.getItemByType(sourceType);
if (null != sparkDataType) {
targetType = sparkDataType.getHiveDataType().getType();
//时间戳字段强转成string字段
if(targetType.equals("timestamps")) targetType.equals("string");
} else {
targetType = HiveDataTypeEnum.STRING.getType();
}
dbTableColumns.add(columnName);
dbTableColumnTypeMap.put(columnName, targetType);
if (i != 0) {
createBuilder.append(",");
}
createBuilder.append(columnName).append(" ").append(targetType);
}
createBuilder.append(") PARTITIONED by (").append(partitionColumn)
.append(" STRING) ");
sparkSession.sql(createTableSQL);

复制代码

1
2
3
4
5
6

2.3 对比字段
我们在2.2中，如果hive有字段了，那么就不会创建表。
问题在于，如果hive中的字段比mysql中的少怎么办？
2.3.1 获取hive中的表字段

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

HiveUtil connectionToHive = new HiveUtil("org.apache.hive.jdbc.HiveDriver", hiveUrl, hiveUser, hivePassword);
public List<String> getTableColumns(String dbName,String tableName) throws SQLException {
ResultSet rs = null;
try {
if (!this.validateTableExist(tableName)) {
return null;
}
DatabaseMetaData metaData = connection.getMetaData();
rs = metaData.getColumns(null, dbName, tableName.toUpperCase(), "%");
List<String> columns = new ArrayList();
while (rs.next()) {
columns.add(rs.getString("COLUMN_NAME").toLowerCase());
}
return columns;
} catch (SQLException e) {
throw e;
} finally {
if (null != rs) {
rs.close();
}
}
}

复制代码

1
2
3

2.3.2 对比字段并且添加：

复制代码

1
2
3
4
5
6
7
8
9
10
11
12

for (String dbTableColumn : dbTableColumns) {
if (StringUtil.hasCapital(dbTableColumn)) {
DingDingAlert.sendMsg(dbTableName + "的" + dbTableColumn + "是大写字段，替换成小写");
logger.warn(dbTableName + "的" + dbTableColumn + "是大写的，把他替换成小写");
sb.append("n " + GetTime.getTimeStamp("yyyy-MM-dd HH:mm:ss") + "| WARN |" + "表" + hiveTableName + "在hive中不存在，程序关闭");
dbTableColumn = StringUtil.convertStringToLowerCase(dbTableColumn, false);
}
if (!hiveTableColumns.contains(dbTableColumn)) {
alterColumns.add(dbTableColumn);
}
}

复制代码

1
2
3

2.4 将内存中的表存入hive

复制代码

1
2
3
4
5

bizdateDS.createOrReplaceTempView(tmpTableName); //注意这里不是直接从mysql抽到hive，而是先从Mysql抽到内存中
insert hive_table
select hive中的已经有的表的字段
from tmpTableName

```