tonglin0325的个人主页

Spark学习笔记——使用CatalystSqlParser解析Spark SQL

Spark的parser使用的是antlr来实现,其g4文件如下

1
2
https://github.com/apache/spark/blob/master/sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4

如果想解析spark SQL的语句,可以使用其原生的parser来进行解析,代码如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
package com.bigdata.spark

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
import org.apache.spark.sql.catalyst.plans.logical.{InsertIntoTable, LogicalPlan}
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.execution.datasources.CreateTable

object SparkSQLParser {

def main(args: Array[String]): Unit = {
// val spark = SparkSession.builder()
// .appName("SQL Create Table Parser")
// .master("local[*]")
// .enableHiveSupport() // 关键:启用 Hive 语法支持
// .getOrCreate()
// val logicalPlan: LogicalPlan = spark.sessionState.sqlParser.parsePlan(sql)

val sql = "SELECT id, name FROM users WHERE age > 18" // select
// val sql = "CREATE TABLE users (id INT, name STRING)" // create
// val sql = "INSERT INTO users VALUES (1, 'Alice')" // insert
// val sql = "INSERT INTO xx.table2 SELECT * FROM xx.table1" // insert


val logicalPlan: LogicalPlan = CatalystSqlParser.parsePlan(sql)
println(logicalPlan)
logicalPlan match {
// 解析建表语句
case createTable: CreateTable =>
println(s"SQL: [$sql] -> 这是一个 CREATE TABLE 语句")
// 解析insert语句
case _: InsertIntoTable =>
println(s"SQL: [$sql] -> 这是一个 INSERT 语句")
// 解析血缘
var inputTables = Set[String]()
var outputTable: Option[String] = None
// 遍历 LogicalPlan 解析血缘
logicalPlan.foreach {
case UnresolvedRelation(tableIdentifier) =>
inputTables += tableIdentifier.quotedString // 解析输入表
case InsertIntoTable(table, _, _, _, _) =>
table match {
case UnresolvedRelation(tableIdentifier) =>
outputTable = Some(tableIdentifier.quotedString) // 解析输出表
case _ =>
outputTable = Some(table.toString()) // 其他情况
}
case _ => // 其他情况忽略
}
println(s"输入表: ${inputTables.mkString(", ")}")
println(s"输出表: ${outputTable.getOrElse("无")}")
// 解析select语句
case _: Project | _: Filter | _: Aggregate | _: Join | _: LogicalPlan =>
println(s"SQL: [$sql] -> 这是一个 SELECT 查询语句")
case _ =>
println(s"SQL: [$sql] -> 未知类型")
}
}

}

1.解析insert语句

1
2
3
4
5
6
7
8
'InsertIntoTable 'UnresolvedRelation `xx`.`table2`, false, false
+- 'Project [*]
+- 'UnresolvedRelation `xx`.`table1`

SQL: [INSERT INTO xx.table2 SELECT * FROM xx.table1] -> 这是一个 INSERT 语句
输入表: `xx`.`table1`
输出表: `xx`.`table2`

2.解析select语句

1
2
3
4
5
6
'Project ['id, 'name]
+- 'Filter ('age > 18)
+- 'UnresolvedRelation `users`

SQL: [SELECT id, name FROM users WHERE age > 18] -> 这是一个 SELECT 查询语句

3.解析create语句

由于spark默认是不支持解析create sql的,需要依赖hive

使用sparksession解析会报

1
2
3
4
5
Exception in thread "main" java.lang.IllegalArgumentException: Unable to instantiate SparkSession with Hive support because Hive classes are not found.
at org.apache.spark.sql.SparkSession$Builder.enableHiveSupport(SparkSession.scala:869)
at com.bigdata.spark.SparkSQLParser$.main(SparkSQLParser.scala:17)
at com.bigdata.spark.SparkSQLParser.main(SparkSQLParser.scala)

使用CatalystSqlParser解析会报

全文 >>

Android常用命令

1.查看Android手机的CPU架构#

1
2
3
$ getprop ro.product.cpu.abi
arm64-v8a

cpu架构表格

架构 Tag 说明
armeabi linux/arm/v5或者linux/arm/v6 第5代第6代的ARM处理器,早期手机用的比较多。比如ARM9,ARM11系列(08年左右的手机,如HTC Hero,使用骁龙MSM7200A处理器,基于 ARM11 架构)
armeabi-v7a linux/arm/v7 第七代及以上的ARM处理器。比如骁龙400系列(12年左右的手机)。
arm64-v8a linux/arm64 第8代64位ARM处理器。比如 ``2016年 之后中高端的手机,比如 骁龙8系列麒麟9系列联发科1000+ 等。
x86 linux/386 32位的桌面PC电脑CPU。平板模拟器用得比较多。
x86_64 linux/amd64 64位桌面PC电脑CPU。64 位的平板。

 

2.termux更换源#

1
2
termux-change-repo

参考:https://mirrors.tuna.tsinghua.edu.cn/help/termux/

YARN学习笔记——调度器

YARN有3种调度器:FIFO Scheduler,Capacity Scheduler(容器调度器) 和 FairS cheduler(公平调度器)。cloudera官方推荐使用Capacity Scheduler,而且在cloudera的CDP中,只保留了Capacity Scheduler这一种调度器。

1.FIFO Scheduler:FIFO Scheduler把应用按提交的顺序排成一个队列,这是一个先进先出队列,在进行资源分配的时候,先给队列中最头上的应用进行分配资源,待最头上的应用需求满足后再给下一个分配,以此类推。在FIFO 调度器中,小任务会被大任务阻塞。

2.Capacity Scheduler(容量调度器)而对于Capacity调度器,有一个专门的队列用来运行小任务,但是为小任务专门设置一个队列会预先占用一定的集群资源,这就导致大任务的执行时间会落后于使用FIFO调度器时的时间。

  1. 多队列:每个队列可配置一定的资源量,每个队列采用FIFO调度策略。
  2. 容量保证:管理员可为每个队列设置资源最低保证和资源使用上限。
  3. 灵活性:如果一个队列中的资源有剩余,可以暂时共享给那些需要资源的队列,而一旦该队列有新的应用 程序提交,则其他队列借调的资源会归还给该队列。
  4. 多租户:支持多用户共享集群(如图中queueC,配置租户ss、cls)和多应用程序同时运行;为了防止同一个用户的作业独占队列中的资源,该调度器会对同一用户提交的作业所占资源量进行限定。

参考:hadoop3 Yarn容量(Capacity Scheduler)调度器和公平(Fair Scheduler)调度器配置

 

 

3.Fair Scheduler(公平调度器)

在Fair调度器中,我们不需要预先占用一定的系统资源,Fair调度器会为所有运行的job动态的调整系统资源。

如下图所示,当第一个大job提交时,只有这一个job在运行,此时它获得了所有集群资源;当第二个小任务提交后,Fair调度器会分配一半资源给这个小任务,让这两个任务公平的共享集群资源。

需要注意的是,在下图Fair调度器中,从第二个任务提交到获得资源会有一定的延迟,因为它需要等待第一个任务释放占用的Container。小任务执行完成之后也会释放自己占用的资源,大任务又获得了全部的系统资源。最终的效果就是Fair调度器即得到了高的资源利用率又能保证小任务及时完成。

全文 >>

mac挂载linux的磁盘

到了2022年,已经不能使用brew安装sshfs来挂载linux的磁盘了,但是可以通过安装包的方式来安装,会报错

1
2
error: sshfs has been disabled because it requires closed-source macFUSE!

首先需要安装sshfs,版本是SSHFS 2.5.0

1
2
https://osxfuse.github.io/2014/02/03/SSHFS-2.5.0.html

其次需要安装macFUSE

1
2
https://osxfuse.github.io/2022/08/19/macFUSE-4.4.1.html

然后就能挂载linux的硬盘了

首先创建目录

1
2
mkdir /Users/xxx/data01

挂载linux的/data01目录到刚刚创建的目录

全文 >>

Linux学习笔记

1.重启或者关机

1
2
shutdown -h now  #立刻进行关机
shutdown -r now或者reboor  #现在重新启动计算机

2.查看本机的IP地址

1
2
ifconfig -a

ubuntu修改IP地址和网关

参考:ubuntu修改IP地址和网关的方法

3.查看内存磁盘CPU

查看内存

1
2
free -m

查看内存及cpu使用情况的命令

1
2
top

也可以安装htop工具,这样更直观,安装命令如下

1
2
sudo apt-get install htop

安装完后,直接输入命令:htop,就可以看到内存或cpu的使用情况了。

查看磁盘容量

查看磁盘使用情况

全文 >>