在scala中使用spark sql解决特定需求

Posted 2021-04-25 我是攻城师

接着上篇文章，本篇来看下如何在scala中完成使用spark sql将不同日期的数据导入不同的es索引里面。

首下看下用到的依赖包有哪些：

下面看相关的代码，代码可直接在跑在win上的idea中，使用的是local模式，数据是模拟造的：

分析下，代码执行过程：

（1）首先创建了一个SparkSession对象，注意这是新版本的写法，然后加入了es相关配置

（2）导入了隐式转化的es相关的包

（3）通过Seq+Tuple创建了一个DataFrame对象，并注册成一个表

（4）导入spark sql后，执行了一个sql分组查询

（5）获取每一组的数据

（6）处理组内的Struct结构

（7）将组内的Seq[Row]转换为rdd，最终转化为df

（8）执行导入es的方法，按天插入不同的索引里面

（9）结束

需要注意的是必须在执行collect方法后，才能在循环内使用sparkContext，否则会报错的，在服务端是不能使用sparkContext的，只有在Driver端才可以。