spaksql把hive导入mysql,SparkSql实现Mysql到hive的数据流动

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

properties.put("user", dbUser);

properties.put("password", dbPassword);

properties.put("driver", dbDriver);

Dataset bizdateDS &＃61; sparkSession.read().jdbc(

dbUrl,

dbTableName,

properties

);

其中&＃xff1a;org.apache.spark.sql.Dataset(这里面试官问我怎么把mysql的数据转化到Spark&＃xff0c;我没答上来)

第二步&＃xff1a;创建数据库与表 2.1 创建库

String createDBSQL &＃61; "CREATE DATABASE IF NOT EXISTS " &＃43; hiveDBName &＃43; " LOCATION &＃39;" &＃43; dbPath &＃43; "&＃39;";

sparkSession.sql(createDBSQL);

&＃96;&＃96;&＃96;

2.2创建表

分成两步&＃xff0c;第一步读取Mysql元数据字段&＃xff0c;第二步把这些字段创建出来

2.2.1 读取mysql字段

StructType structType &＃61; bizdateDS.schema();

StructField[] structFields &＃61; structType.fields();

/*

structField是StructType中的字段。

param&＃xff1a;name此字段的名称。

param&＃xff1a;dataType此字段的数据类型。

param&＃xff1a;nullable指示此字段的值是否为空值。

param&＃xff1a;metadata此字段的元数据。 如果未修改列的内容(例如&＃xff0c;在选择中)&＃xff0c;则应在转换期间保留元数据。

*/

2.2.2 创建字段

String sourceType; //Name of the type used in JSON serialization.

String columnName;

String targetType;

StructField structField;

SparkDataTypeEnum sparkDataType;

StringBuilder createBuilder &＃61; new StringBuilder(capacity);

createBuilder.append("CREATE TABLE IF NOT EXISTS ").append(realHiveTableName).append(" (");

List dbTableColumns &＃61; Lists.newArrayList();

Map dbTableColumnTypeMap &＃61; Maps.newHashMap();

//把Mysql中的每个字段都提取出来

for (int i &＃61; 0, len &＃61; structFields.length; i

structField &＃61; structFields[i];

sourceType &＃61; structField.dataType().typeName();

columnName &＃61; structField.name();

if (sourceType.contains("(")) { //处理类似varchar(20)

sourceType &＃61; sourceType.substring(0, sourceType.indexOf("("));

}

sparkDataType &＃61; SparkDataTypeEnum.getItemByType(sourceType);

if (null !&＃61; sparkDataType) {

targetType &＃61; sparkDataType.getHiveDataType().getType();

//时间戳字段强转成string字段

if(targetType.equals("timestamps")) targetType.equals("string");

} else {

targetType &＃61; HiveDataTypeEnum.STRING.getType();

}

dbTableColumns.add(columnName);

dbTableColumnTypeMap.put(columnName, targetType);

if (i !&＃61; 0) {

createBuilder.append(",");

}

createBuilder.append(columnName).append(" ").append(targetType);

}

createBuilder.append(") PARTITIONED by (").append(partitionColumn)

.append(" STRING) ");

sparkSession.sql(createTableSQL);

2.3 对比字段

我们在2.2中&＃xff0c;如果hive有字段了&＃xff0c;那么就不会创建表。

问题在于&＃xff0c;如果hive中的字段比mysql中的少怎么办&＃xff1f;

2.3.1 获取hive中的表字段

HiveUtil connectionToHive &＃61; new HiveUtil("org.apache.hive.jdbc.HiveDriver", hiveUrl, hiveUser, hivePassword);

public List getTableColumns(String dbName,String tableName) throws SQLException {

ResultSet rs &＃61; null;

try {

if (!this.validateTableExist(tableName)) {

return null;

}

DatabaseMetaData metaData &＃61; connection.getMetaData();

rs &＃61; metaData.getColumns(null, dbName, tableName.toUpperCase(), "%");

List columns &＃61; new ArrayList();

while (rs.next()) {

columns.add(rs.getString("COLUMN_NAME").toLowerCase());

}

return columns;

} catch (SQLException e) {

throw e;

} finally {

if (null !&＃61; rs) {

rs.close();

}

}

}

2.3.2 对比字段并且添加&＃xff1a;

for (String dbTableColumn : dbTableColumns) {

if (StringUtil.hasCapital(dbTableColumn)) {

DingDingAlert.sendMsg(dbTableName &＃43; "的" &＃43; dbTableColumn &＃43; "是大写字段&＃xff0c;替换成小写");

logger.warn(dbTableName &＃43; "的" &＃43; dbTableColumn &＃43; "是大写的&＃xff0c;把他替换成小写");

sb.append("\n " &＃43; GetTime.getTimeStamp("yyyy-MM-dd HH:mm:ss") &＃43; "| WARN |" &＃43; "表" &＃43; hiveTableName &＃43; "在hive中不存在&＃xff0c;程序关闭");

dbTableColumn &＃61; StringUtil.convertStringToLowerCase(dbTableColumn, false);

}

if (!hiveTableColumns.contains(dbTableColumn)) {

alterColumns.add(dbTableColumn);

}

}

2.4 将内存中的表存入hive

bizdateDS.createOrReplaceTempView(tmpTableName); //注意这里不是直接从mysql抽到hive&＃xff0c;而是先从Mysql抽到内存中

insert hive_table select hive中的已经有的表的字段 from tmpTableName

##很明显的&＃xff0c;如果不是需要和hive已经有的表交互根本用不到jdbc