今天去面试了一波,因为调度系统采用了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的数据流动的更多相关文章
-
从MySQL到Hive,数据迁移就这么简单
使用Sqoop能够极大简化MySQL数据迁移至Hive之流程,并降低Hadoop处理分析任务时的难度. 先决条件:安装并运行有Sqoop与Hive的Hadoop环境.为了加快处理速度,我们还将使用Cl ...
-
使用Sqoop从mysql向hdfs或者hive导入数据时出现的一些错误
1.原表没有设置主键,出现错误提示: ERROR tool.ImportTool: Error during import: No primary key could be found for tab ...
-
Spark操作MySQL,Hive并写入MySQL数据库
最近一个项目,需要操作近70亿数据进行统计分析.如果存入MySQL,很难读取如此大的数据,即使使用搜索引擎,也是非常慢.经过调研决定借助我们公司大数据平台结合Spark技术完成这么大数据量的统计分析. ...
-
从hive将数据导出到mysql(转)
从hive将数据导出到mysql http://abloz.com 2012.7.20 author:周海汉 在上一篇文章<用sqoop进行mysql和hdfs系统间的数据互导>中,提到s ...
-
sqoop用法之mysql与hive数据导入导出
目录 一. Sqoop介绍 二. Mysql 数据导入到 Hive 三. Hive数据导入到Mysql 四. mysql数据增量导入hive 1. 基于递增列Append导入 1). 创建hive表 ...
-
Hive[4] 数据定义 HiveQL
HiveQL 是 Hive 查询语言,它不完全遵守任一种 ANSI SQL 标准的修订版,但它与 MySQL 最接近,但还有显著的差异,Hive 不支持行级插入,更新和删除的操作,也不支持事务,但 H ...
-
hadoop笔记之Hive的数据存储(视图)
Hive的数据存储(视图) Hive的数据存储(视图) 视图(view) 视图是一种虚表,是一个逻辑概念:可以跨越多张表 既然视图是一种虚表,那么也就是说用操作表的方式也可以操作视图 但是视图是建立在 ...
-
Sqoop使用,mysql,hbase,hive等相互转换
Sqoop 是一款用来在不同数据存储软件之间进行数据传输的开源软件,它支持多种类型的数据储存软件. 安装 Sqoop 1.下载sqoop并加mysql驱动包 http://mirror.bit.edu ...
-
使用sqoop将MySQL数据库中的数据导入Hbase
使用sqoop将MySQL数据库中的数据导入Hbase 前提:安装好 sqoop.hbase. 下载jbdc驱动:mysql-connector-java-5.1.10.jar 将 mysql-con ...
随机推荐
-
java守护线程的理解
package daemonThread; /*setDaemon(true)方法将线程设置为守护线程,线程的Daemon默认值为false * 只要当前JVM实例中存在任何一个非守护线程没有结束,守 ...
-
7 static关键字
class Person { static int i; static void fun() { System.out.println("我是静态函数"); } } 在静态函数中不 ...
-
再看GOPATH
原本不打算介绍GOPATH,然而,总是有初学者问一些关于GOPATH的问题,因此在这里再介绍一下GOPATH GOPATH环境变量用于指定这样一些目录:除$GOROOT之外的包含Go项目源代码和二进制 ...
-
【原创】纯OO:从设计到编码写一个FlappyBird (四)
第三部分请点这里 这里来实现Obstacle类.其实flappybird的本质就是小鸟原地掉,然后几根柱子在走.这也是在Game类里,用obs.move()来实现游戏逻辑的原因. 我们首先必须确定几个 ...
-
Jquery源码分析与简单模拟实现
前言 最近学习了一下jQuery源码,顺便总结一下,版本:v2.0.3 主要是通过简单模拟实现jQuery的封装/调用.选择器.类级别扩展等.加深对js/Jquery的理解. 正文 先来说问题: 1. ...
-
[Swift]LeetCode162. 寻找峰值 | Find Peak Element
A peak element is an element that is greater than its neighbors. Given an input array nums, where nu ...
-
jenkins shell部署
jenkins执行shell脚本 jenkins执行shell 上一篇说的是jenkins+svn+maven把war包自动部署到Tomcat,这篇是从SVN上拉取代码maven生成jar文件,并且拷 ...
-
BUPT2017 wintertraining(15) #2 题解
这场有点难,QAQ.补了好久(。• ︿•̀。) ,总算能写题解了(つд⊂) A. Beautiful numbers CodeForces - 55D 题意 求\([l,r](1\le l_i\l ...
-
Linux proc目录下 几个系统文件下的各项参数 (cpuinfo,uptime,meminfo,stat,loadavg)
参考链接: Linux 操作系统内核基本实验.pdf http://max.book118.com/html/2015/0919/25787869.shtm Linux下cpuinfo文件各项参数的详 ...
-
第三百九十二节,Django+Xadmin打造上线标准的在线教育平台—sql注入攻击,xss攻击,csrf攻击
第三百九十二节,Django+Xadmin打造上线标准的在线教育平台—sql注入攻击,xss攻击,csrf攻击 sql注入攻击 也就是黑客通过表单提交的地方,在表单里输入了sql语句,就是通过SQL语 ...