SparkSql实现Mysql到hive的数据流动

时间:2021-05-21 13:07:16

今天去面试了一波,因为调度系统采用了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

SparkSql实现Mysql到hive的数据流动的更多相关文章

  1. 从MySQL到Hive,数据迁移就这么简单

    使用Sqoop能够极大简化MySQL数据迁移至Hive之流程,并降低Hadoop处理分析任务时的难度. 先决条件:安装并运行有Sqoop与Hive的Hadoop环境.为了加快处理速度,我们还将使用Cl ...

  2. 使用Sqoop从mysql向hdfs或者hive导入数据时出现的一些错误

    1.原表没有设置主键,出现错误提示: ERROR tool.ImportTool: Error during import: No primary key could be found for tab ...

  3. Spark操作MySQL,Hive并写入MySQL数据库

    最近一个项目,需要操作近70亿数据进行统计分析.如果存入MySQL,很难读取如此大的数据,即使使用搜索引擎,也是非常慢.经过调研决定借助我们公司大数据平台结合Spark技术完成这么大数据量的统计分析. ...

  4. 从hive将数据导出到mysql(转)

    从hive将数据导出到mysql http://abloz.com 2012.7.20 author:周海汉 在上一篇文章<用sqoop进行mysql和hdfs系统间的数据互导>中,提到s ...

  5. sqoop用法之mysql与hive数据导入导出

    目录 一. Sqoop介绍 二. Mysql 数据导入到 Hive 三. Hive数据导入到Mysql 四. mysql数据增量导入hive 1. 基于递增列Append导入 1). 创建hive表 ...

  6. Hive&lbrack;4&rsqb; 数据定义 HiveQL

    HiveQL 是 Hive 查询语言,它不完全遵守任一种 ANSI SQL 标准的修订版,但它与 MySQL 最接近,但还有显著的差异,Hive 不支持行级插入,更新和删除的操作,也不支持事务,但 H ...

  7. hadoop笔记之Hive的数据存储&lpar;视图&rpar;

    Hive的数据存储(视图) Hive的数据存储(视图) 视图(view) 视图是一种虚表,是一个逻辑概念:可以跨越多张表 既然视图是一种虚表,那么也就是说用操作表的方式也可以操作视图 但是视图是建立在 ...

  8. Sqoop使用,mysql&comma;hbase&comma;hive等相互转换

    Sqoop 是一款用来在不同数据存储软件之间进行数据传输的开源软件,它支持多种类型的数据储存软件. 安装 Sqoop 1.下载sqoop并加mysql驱动包 http://mirror.bit.edu ...

  9. 使用sqoop将MySQL数据库中的数据导入Hbase

    使用sqoop将MySQL数据库中的数据导入Hbase 前提:安装好 sqoop.hbase. 下载jbdc驱动:mysql-connector-java-5.1.10.jar 将 mysql-con ...

随机推荐

  1. java守护线程的理解

    package daemonThread; /*setDaemon(true)方法将线程设置为守护线程,线程的Daemon默认值为false * 只要当前JVM实例中存在任何一个非守护线程没有结束,守 ...

  2. 7 static关键字

    class Person { static int i; static void fun() { System.out.println("我是静态函数"); } } 在静态函数中不 ...

  3. 再看GOPATH

    原本不打算介绍GOPATH,然而,总是有初学者问一些关于GOPATH的问题,因此在这里再介绍一下GOPATH GOPATH环境变量用于指定这样一些目录:除$GOROOT之外的包含Go项目源代码和二进制 ...

  4. 【原创】纯OO:从设计到编码写一个FlappyBird (四)

    第三部分请点这里 这里来实现Obstacle类.其实flappybird的本质就是小鸟原地掉,然后几根柱子在走.这也是在Game类里,用obs.move()来实现游戏逻辑的原因. 我们首先必须确定几个 ...

  5. Jquery源码分析与简单模拟实现

    前言 最近学习了一下jQuery源码,顺便总结一下,版本:v2.0.3 主要是通过简单模拟实现jQuery的封装/调用.选择器.类级别扩展等.加深对js/Jquery的理解. 正文 先来说问题: 1. ...

  6. &lbrack;Swift&rsqb;LeetCode162&period; 寻找峰值 &vert; Find Peak Element

    A peak element is an element that is greater than its neighbors. Given an input array nums, where nu ...

  7. jenkins shell部署

    jenkins执行shell脚本 jenkins执行shell 上一篇说的是jenkins+svn+maven把war包自动部署到Tomcat,这篇是从SVN上拉取代码maven生成jar文件,并且拷 ...

  8. BUPT2017 wintertraining&lpar;15&rpar; &num;2 题解

    这场有点难,QAQ.补了好久(。• ︿•̀。) ,总算能写题解了(つд⊂) A. Beautiful numbers CodeForces - 55D 题意 ​ 求\([l,r](1\le l_i\l ...

  9. Linux proc目录下 几个系统文件下的各项参数 (cpuinfo,uptime,meminfo,stat,loadavg)

    参考链接: Linux 操作系统内核基本实验.pdf http://max.book118.com/html/2015/0919/25787869.shtm Linux下cpuinfo文件各项参数的详 ...

  10. 第三百九十二节,Django&plus;Xadmin打造上线标准的在线教育平台—sql注入攻击&comma;xss攻击&comma;csrf攻击

    第三百九十二节,Django+Xadmin打造上线标准的在线教育平台—sql注入攻击,xss攻击,csrf攻击 sql注入攻击 也就是黑客通过表单提交的地方,在表单里输入了sql语句,就是通过SQL语 ...