当前位置:网站首页>Spark SQL UDF function
Spark SQL UDF function
2022-07-25 15:15:00 【The south wind knows what I mean】
UDF function
def main(args: Array[String]): Unit = {
val session: SparkSession = SparkSession
.builder()
.config("dfs.ha.namenodes.nameservices","xxxxx")
.appName("sql demo")
.master("local")
.enableHiveSupport()
.getOrCreate()
// Custom function
val code: UserDefinedFunction = session.udf.register("code", (arg: Int) => {
if (arg > 5) "F" else "M"
})
val sql_1:String =
s""" |select * from |sparktuning.course_pay |""".stripMargin
val df: DataFrame = session.sql(sql_1)
// usage 1.
df.select("chapterid")
.withColumn("flag",code(col("chapterid")))
.withColumnRenamed("chapterid","new_chapterid")
.show(false)
+-------------+----+
|new_chapterid|flag|
+-------------+----+
|4 |M |
|7 |F |
|8 |F |
|5 |M |
|4 |M |
|9 |F |
|7 |F |
|0 |M |
|5 |M |
|4 |M |
|4 |M |
|0 |M |
|1 |M |
|4 |M |
|9 |F |
+-------------+----+
// usage 2.
val sql_2:String =
s""" |select | code(chapterid) Sex |from |sparktuning.course_pay |""".stripMargin
session.sql(sql_2).show(false)
+---+
|Sex|
+---+
|M |
|F |
|F |
|M |
|M |
|F |
|F |
|M |
|M |
|M |
|M |
|M |
|M |
|M |
|F |
+---+
}
边栏推荐
- Spark002 --- spark task submission, pass JSON as a parameter
- 反射-笔记
- Share a department design method that avoids recursion
- Iframe nested other website page full screen settings
- Recommend 10 learning websites that can be called artifact
- 瀑布流布局
- When using jetty to run items, an error is reported: form too large or form too many keys
- Sublimetext-win10 cursor following problem
- js URLEncode函数
- Leo-sam: tightly coupled laser inertial odometer with smoothing and mapping
猜你喜欢

记一次Yarn Required executor memeory is above the max threshold(8192MB) of this cluster!

Process control (Part 1)

Single or multiple human posture estimation using openpose
[Android] recyclerview caching mechanism, is it really difficult to understand? What level of cache is it?

Nacos2.1.0 cluster construction

Use the command to check the WiFi connection password under win10 system

ice 100G 网卡分片报文 hash 问题

Debounce and throttle

System. Accessviolationexception: an attempt was made to read or write to protected memory. This usually indicates that other memory is corrupted

Client error: invalid param endpoint is blank
随机推荐
请问seata中mysql参数每个客户端连接最大的错误允许数量要怎么理解呢?
Scala111-map、flatten、flatMap
解决asp.net上传文件时文件太大导致的错误
mysql heap表_MySQL内存表heap使用总结-九五小庞
TypeScript学习2——接口
Raft of distributed consistency protocol
RedisCluster搭建和扩容
什么是物联网
Promise object and macro task, micro task
Deployment and simple use of PostgreSQL learning
Pl/sql creates and executes ORALCE stored procedures and returns the result set
流程控制(上)
pkg_resources动态加载插件
What is the Internet of things
如何更新更新数据库中的json值?
延迟加载源码剖析:
Introduction to raspberry Pie: initial settings of raspberry pie
Spark AQE
bridge-nf-call-ip6tables is an unknown key异常处理
ESXI6.7.0 升级到7.0U3f(2022年7月12 更新)