SparkSql實現Mysql到hive的數據流動

今天去面試了一波，因為調度系統采用了SparkSql實現數據從Mysql到hive，在這一點上面試官很明顯很不滿我對於Spark的理解，19年的第一個面試就這么掛了。
有問題不怕，怕的是知道了問題還得過且過。現在就來梳理下我的項目是怎么使用Spark導數的
第一步：把mysql中的表放入內存

        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 創建庫

        String createDBSQL = "CREATE DATABASE IF NOT EXISTS " + hiveDBName + " LOCATION '" + dbPath + "'";
        sparkSession.sql(createDBSQL);
      ```  
        2.2創建表
        分成兩步，第一步讀取Mysql元數據字段，第二步把這些字段創建出來
        2.2.1 讀取mysql字段

 StructType structType = bizdateDS.schema();
    StructField[] structFields = structType.fields();
        /*
        structField是StructType中的字段。
        param：name此字段的名稱。
        param：dataType此字段的數據類型。
        param：nullable指示此字段的值是否為空值。
        param：metadata此字段的元數據。 如果未修改列的內容（例如，在選擇中），則應在轉換期間保留元數據。
         */

     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);

2.3 對比字段
    我們在2.2中，如果hive有字段了，那么就不會創建表。
    問題在於，如果hive中的字段比mysql中的少怎么辦？        
 2.3.1 獲取hive中的表字段

     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();
        }
    }
}

2.3.2 對比字段並且添加：

    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);
        }
    }

2.4 將內存中的表存入hive

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

##很明顯的，如果不是需要和hive已經有的表交互根本用不到jdbc