大数据中spark跟sparksql写入es数据库

作者：小涛

object Legend01 {

private val logger: Logger = LoggerFactory.getLogger(“Legend”)

def main(args: Array[String]): Unit = {

if (args.length != 1) {

println(
“”"
|请输入applist数据的输入路径：
|input:
“”".stripMargin)
sys.exit(1)
}
//用数组接收所有数据的输入路径
val Array(input) = args

val session = SparkSession.builder() //初始化spark对象
.master(“local[*]”)
.config(“spark.testing.memory”, “2147480000”)
.appName(Legend01.getClass.getSimpleName)
.config(“spark.serializer”, “org.apache.spark.serializer.KryoSerializer”)
.config(“es.nodes”, “ip”)
.config(“es.port”, “9200”)
.config(“es.index.auto.create”, “true”)
.getOrCreate()

//读取applist数据并且去重
var Get = session.read.textFile(input).distinct()

//导入隐士转换
import session.implicits._
val jsonj = Get.map({ t =>

val data = null

try {

val nObject = JSON.parseObject(t)

val imei = nObject.getString(“imei”)

val imsi: String = nObject.getString(“imsi”)

val extras = nObject.getJSONObject(“extras”)

val app_list = extras.getJSONArray(“app_list”)

val trim = app_list.toString.trim

val strings = trim.substring(1, trim.length - 1).split(",")

val result = new StringBuilder

val sb = new StringBuilder

for (i <- strings) {

val quchukongge = i.trim

val str = quchukongge.substring(1, quchukongge.length - 1)

sb.append(str + “,”)

}

result.append(imei + “,” + imsi + “,” + sb).toString()

} catch {
case e: Exception => {

//如果数据不规范跟不是合法数据进行抛异常
logger.error("数据为不合法数据请检查不合法数据是否有用 : " + t)

}
data
}
}
)

val dewq = jsonj.filter(b => b != null).cache()

val mj = dewq.rdd.filter(b => b != null).cache()
val dewr = mj.filter(x => {

// x.contains(“com.aifeng.woshichuanqi.yingyongbao”) || x.contains(“com.tencent.cqsj”) || x.contains(“com.tencent.tmgp.rxcq”) || x.contains(“com.tencent.tmgp.shenghewzcq”) || x.contains(“com.tencent.tmgp.xymobile”) || x.contains(“com.tmcq.anzhi”)

x.contains(“金融”) || x.contains(“贷款”) || x.contains(“信用卡”) || x.contains(“com.hejgjs.shiyuantouzi”) || x.contains(“net.metaquotes.metatrader4”) || x.contains(“disk.micro.com.microdisk”) || x.contains(“com.pme91.guoxin.app.android”) || x.contains(“com.gwtsz.gts2.bj”) || x.contains(“com.pme91.app.android”) || x.contains(“io.dcloud.com.FKqihuo.app”) || x.contains(“com.qhdhcd.cn”) || x.contains(“com.gwtsz.gts2mj1.hx”) || x.contains(“com.baidao.ytxmobile”) || x.contains(“com.jin91.baiyintong”) || x.contains(“com.wattforex”)

})
val Arrt: RDD[(String, String)] = dewr.map(x => {

val de: Array[String] = x.split(",")

val da: String = de(0)

val da01: String = de(1)

(da, da01)
}
)

case class Ao(

imei:String,

imsi:String

)

Arrt.map(
x =>
Ao(
x._1,
x._2
)
)

.saveToEs(“sparktes/test02”)

大数据中spark跟sparksql写入es数据库相关推荐

【Spark Summit East 2017】管道泄漏问题：像女士一样在大数据中做个的标记
更多精彩内容参见云栖社区大数据频道https://yq.aliyun.com/big-data:此外,通过Maxcompute及其配套产品,低廉的大数据分析仅需几步,详情访问https://www.a ...
【讲稿】Impala在网易大数据中使用和优化实践-P2
本文是在2020 DataFunCon上所做分享的讲稿,感谢DataFun团队整理.由于篇幅较长,分为2篇来记录,这是第二篇. 上文:[讲稿]Impala在网易大数据中使用和优化实践-P1 3.基于Z ...
尚硅谷大数据技术Spark教程-笔记01【Spark(概述、快速上手、运行环境、运行架构)】
视频地址:尚硅谷大数据Spark教程从入门到精通_哔哩哔哩_bilibili 尚硅谷大数据技术Spark教程-笔记01[Spark(概述.快速上手.运行环境.运行架构)] 尚硅谷大数据技术Spark教 ...
尚硅谷大数据技术Spark教程-笔记09【SparkStreaming（概念、入门、DStream入门、案例实操、总结）】
尚硅谷大数据技术-教程-学习路线-笔记汇总表[课程资料下载] 视频地址:尚硅谷大数据Spark教程从入门到精通_哔哩哔哩_bilibili 尚硅谷大数据技术Spark教程-笔记01[SparkCore ...
7、大数据中常见的文件存储格式以及hadoop中支持的压缩算法
Hadoop系列文章目录 1.hadoop3.1.4简单介绍及部署.简单验证 2.HDFS操作 - shell客户端 3.HDFS的使用(读写.上传.下载.遍历.查找文件.整个目录拷贝.只拷贝文件.列 ...
大数据之spark学习记录二: Spark的安装与上手
大数据之spark学习记录二: Spark的安装与上手文章目录大数据之spark学习记录二: Spark的安装与上手 Spark安装本地模式 Standalone 模式基本配置步骤1: 复制 ...
大数据篇--Spark常见面试题总结一
文章目录一.Spark 概念.模块 1.相关概念: 2.基本模块: 二.Spark作业提交流程是怎么样的三.Spark on YARN两种方式的区别以及工作流程 1.Yarn组件简介: 2.Spa ...
Apache Kylin 在小米大数据中的应用
在近期的 Apache Kylin Meetup 北京站上,我们邀请到小米大数据平台 OLAP 负责人陈学辉分享了 Kylin 在小米大数据中的应用. △陈学辉小米拥有众多产品线,业务遍及全球 80 ...
大数据中常见的端口号总结汇总大全（最新）
大数据中常见的端口号总结大全如图所示: 常见端口汇总: Hadoop: 50070:HDFS WEB UI端口 8020 : 高可用的HDFS RPC端口 9000 : 非高可用的HDFS RPC ...

大数据中spark跟sparksql写入es数据库

大数据中spark跟sparksql写入es数据库相关推荐

最新文章

热门文章