내 연락처 정보
우편메소피아@프로톤메일.com
2024-07-12
한어Русский языкEnglishFrançaisIndonesianSanskrit日本語DeutschPortuguêsΕλληνικάespañolItalianoSuomalainenLatina
스파크 SQL 구조화된 데이터를 처리하는 데 특별히 사용되는 Apache Spark의 모듈입니다. SQL 쿼리와 Spark 프로그래밍의 강력한 기능을 통합하여 빅 데이터를 보다 효율적이고 쉽게 처리할 수 있습니다. Spark SQL을 통해 사용자는 Spark에서 직접 SQL 쿼리를 사용하거나 데이터 작업에 DataFrame 및 DataSet API를 사용할 수 있습니다.
Spark SQL의 아키텍처는 주로 다음 구성 요소로 구성됩니다.
쿼리 구문 분석: SQL 쿼리를 추상 구문 트리(AST)로 구문 분석합니다.
논리적 계획 생성:AST를 최적화되지 않은 논리적 계획으로 변환합니다.
논리적 계획 최적화: Catalyst 최적화 프로그램을 사용하여 논리적 계획에 대한 일련의 규칙을 최적화합니다.
물리적 계획 생성: 최적화된 논리적 계획을 하나 이상의 물리적 계획으로 변환하고 최적의 물리적 계획을 선택합니다.
실행: 물리적 계획을 RDD로 변환하고 클러스터에서 병렬로 실행합니다.
스파크컨텍스트 : SparkContext는 Spark 애플리케이션의 주요 진입점이며 Spark 클러스터 연결, 리소스 관리 및 작업 예약을 담당합니다. Spark 2.0 이후에는 SparkContext 대신 SparkSession을 사용하는 것이 좋습니다.
SQL 컨텍스트 : SQLContext는 Spark SQL의 프로그래밍 진입점으로, 사용자가 SQL 쿼리 또는 DataFrame API를 통해 데이터 처리를 수행할 수 있습니다. 기본적인 Spark SQL 기능을 제공합니다.
하이브 컨텍스트: HiveContext는 Hive에 대한 통합 지원을 추가하는 SQLContext의 하위 집합입니다. Hive의 UDF 및 UDAF를 사용하여 Hive의 데이터 및 메타데이터에 직접 액세스할 수 있습니다.
스파크세션 : SparkSession은 Spark 2.0에서 도입된 새로운 개념으로, SQLContext와 HiveContext의 기능을 병합하여 통일된 프로그래밍 인터페이스를 제공합니다. SparkSession은 DataFrame 및 Dataset API를 사용한 데이터 처리를 지원하는 Spark SQL의 권장 진입점입니다.
SparkContext 및 SparkSession 생성에 대한 참고 사항 : SparkContext와 SparkSession을 동시에 생성해야 하는 경우 먼저 SparkContext를 생성한 후 SparkSession을 생성해야 합니다. SparkSession을 먼저 생성한 후 SparkContext를 생성하면 동일한 JVM에서 하나의 SparkContext만 실행할 수 있으므로 예외가 발생합니다.
<properties>
<spark.version>3.1.2</spark.version>
<spark.scala.version>2.12</spark.scala.version>
</properties>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_${spark.scala.version}</artifactId>
<version>${spark.version}</version>
</dependency>
Spark SQL에서 데이터 세트는 주로 DataFrame과 Dataset 유형으로 나뉩니다. 이는 구조화된 데이터와 반구조화된 데이터를 처리하고 조작하기 위한 핵심 추상화입니다.
데이터세트는 Spark 2.0에 도입된 새로운 추상 데이터 구조로, 강력한 형식이며 JVM 개체를 저장할 수 있습니다. Dataset API는 DataFrame의 운영 단순성과 유형 안전성을 결합하며 더 높은 수준의 데이터 유형 제어 및 객체 지향 프로그래밍 스타일이 필요한 시나리오에 적합합니다. 구체적인 특징은 다음과 같습니다.
struct、map 그리고array。데이터세트는 Spark 2.0에 도입된 새로운 추상 데이터 구조로, 강력한 형식이며 JVM 개체를 저장할 수 있습니다. Dataset API는 DataFrame의 운영 단순성과 유형 안전성을 결합하며 더 높은 수준의 데이터 유형 제어 및 객체 지향 프로그래밍 스타일이 필요한 시나리오에 적합합니다. 구체적인 특징은 다음과 같습니다.
DataFrame = Dataset[Row]。import org.apache.spark.sql.SparkSession
object SparkSqlContext {
def main(args: Array[String]): Unit = {
// 创建 SparkConf 对象，设置应用程序的配置
val conf: SparkConf = new SparkConf()
.setMaster("local[4]") // 设置本地运行模式，使用 4 个线程
.setAppName("spark sql") // 设置应用程序名称为 "spark sql"
// 创建 SparkSession 对象，用于 Spark SQL 的编程入口
val spark: SparkSession = SparkSession.builder()
.config(conf) // 将 SparkConf 配置应用于 SparkSession
.getOrCreate() // 获取现有的 SparkSession，或者新建一个
// 获取 SparkContext 对象，可以直接从 SparkSession 中获取
val sc: SparkContext = spark.sparkContext
// 导入 SparkSession 的隐式转换，可以使用 DataFrame API 的方法
import spark.implicits._
// 在这里可以编写数据处理代码，例如创建 DataFrame 和 Dataset，进行数据操作等...
// 停止 SparkSession，释放资源
spark.stop()
}
}
1. 컬렉션에서 만들기
case class Person(name: String, age: Int) // 下同
val data1 = Seq(Person("Alice", 25), Person("Bob", 30))
val ds: Dataset[Person] = spark.createDataset(data) // 这里的spark是SparkSession对象(如上代码)，下同
val data2 = Seq(("Alice", 25), ("Bob", 30))
val df: DataFrame = data.toDF("name", "age")
1. 파일 시스템에서 읽기
val schema = StructType(Seq(
StructField("name", StringType, nullable = false),
StructField("age", IntegerType, nullable = false)
))
val dsJson: Dataset[Person] = spark.read.json("/path/to/json/file").as[Person]
val dfCsv: DataFrame = spark.read
// 使用.schema方法指定CSV文件的模式（schema）其定义了DataFrame的列名和类型。
// 这是一个可选步骤，但如果CSV文件没有头部行，或者你想覆盖文件中的头部行，则必须指定。
.schema(schema)
// 这里设置"header"为"true"，表示CSV文件的第一行是列名，不需要Spark从文件中自动推断。
.option("header", "true")
.csv("/path/to/csv/file")
3. 관계형 데이터베이스에서 읽기
val url = "jdbc:mysql://localhost:3306/database"
val properties = new java.util.Properties()
properties.setProperty("user", "username")
properties.setProperty("password", "password")
val dsDb: Dataset[Person] = spark.read.jdbc(url, "table", properties).as[Person]
val dfDb: DataFrame = spark.read.jdbc(url, "table", properties)
4. 구조화되지 않은 데이터 소스에서 읽기
val dsParquet: Dataset[Person] = spark.read.parquet("/path/to/parquet/file").as[Person]
val dfParquet: DataFrame = spark.read.parquet("/path/to/parquet/file")
5. 수동으로 데이터세트 생성
import org.apache.spark.sql.types._
val schema = StructType(Seq(
StructField("name", StringType, nullable = false),
StructField("age", IntegerType, nullable = false)
))
val data = Seq(Row("Alice", 25), Row("Bob", 30))
val dsManual: Dataset[Person] = spark.createDataFrame(spark.sparkContext.parallelize(data), schema).as[Person]
val dfManual: DataFrame = spark.createDataFrame(
spark.sparkContext.parallelize(data), schema
)
문법 예 1
시뮬레이션 데이터(1000개 항목):
id,name,gender,age,city
1,邵睿,男,12,上海市
2,林子异,男,48,广州市
3,孟秀英,女,46,上海市
4,金嘉伦,男,8,北京市
...
요구 사항: 더 많은 인구(ID 수 >50)와 함께 평균 연령이 가장 높은 도시 및 성별 조합과 각 성별 내에서 이러한 조합의 순위.
// 导入SparkSession的隐式转换，这样可以使用DataFrame的便捷方法(例如下面的'$'符号)
import spark.implicits._
// 定义了一个DataFrame的schema，但在这个例子中，使用了CSV的header来自动推断schema
val schema = StructType(Seq(
StructField("id", LongType),
StructField("name", StringType),
StructField("gender", StringType),
StructField("age", IntegerType),
StructField("city", StringType),
))
// 定义WindowSpec，用于后续的窗口函数操作，按gender分区，按avg_age降序排序，(复用使用此)
val WindowSpec: WindowSpec = Window
.partitionBy($"gender")
.orderBy($"avg_age".desc)
// 从CSV文件中读取数据，使用header作为列名，然后选择特定的列，进行分组和聚合操作
// 哪些城市和性别组合在人口较多（ID数量>50）的情况下具有最高的平均年龄，以及这些组合在各自性别中的排名。
spark.read
// .schema(schema) // 应用我们定义的schema
.option("header", "true") // 使用CSV的header作为列名
.csv("D:\projects\sparkSql\people.csv") // DataFrame
.select($"id", $"name", $"age", $"city", $"gender") // 选择需要的列(不写默认就是全选)
.groupBy($"city", $"gender") // 按城市和性别分组
.agg( // 多重聚合
count($"id").as("count"), // 计算每个组的ID数量
round(avg($"age"), 2).as("avg_age") // 计算每个组的平均年龄，并保留两位小数
)
.where($"count".gt(50)) // 过滤出ID数量大于(可以使用>)50的组
.orderBy($"avg_age".desc) // 按平均年龄降序排序
.select($"city", $"gender", $"avg_age",
dense_rank().over(Window.partitionBy($"gender").orderBy($"avg_age".desc)).as("gender_avg_age_rank"))
.show() // 显示结果
결과:
+------+------+-------+-------------------+
| city|gender|avg_age|gender_avg_age_rank|
+------+------+-------+-------------------+
|北京市| 男| 41.05| 1|
| 东莞| 男| 42.81| 2|
|上海市| 男| 43.92| 3|
|成都市| 男| 45.89| 4|
| 中山| 男| 47.08| 5|
|广州市| 男| 47.47| 6|
| 深圳| 男| 48.36| 7|
|上海市| 女| 46.02| 1|
| 中山| 女| 49.55| 2|
+------+------+-------+-------------------+
구문 예제 2: view, sql
// 读取CSV文件到DataFrame，使用header作为列名
val dfPeople: DataFrame = spark.read
.option("header", "true") // 使用CSV的header作为列名
.csv("D:\projects\sparkSql\people.csv")
// 将DataFrame注册为临时视图
dfPeople.createOrReplaceTempView("people_view")
// 可以使用Spark SQL来查询这个视图了
// 例如，查询所有人的姓名和年龄
spark.sql("SELECT name, age FROM people_view").show()
// 二
spark.sql(
"""
|select * from people_view
|where gender = '男'
|""".stripMargin
).show()
구문 예 3: Join
case class Student(name: String, classId: Int)
case class Class(classId: Int, className: String)
val frmStu = spark.createDataFrame(
Seq(
Student("张三", 1),
Student("李四", 1),
Student("王五", 2),
Student("赵六", 2),
Student("李明", 2),
Student("王刚", 4),
Student("王朋", 5),
)
)
val frmClass = spark.createDataFrame(
Seq(
Class(1, "name1"),
Class(2, "name2"),
Class(3, "name3"),
Class(4, "name4")
)
)
left 왼쪽 조인,rignt 바로 가입,full 완전 외부 조인,anti왼쪽 차이 세트,semi왼쪽 교차로
// 别名 + inner 内连接
frmStu.as("S")
.join(frmClass.as("C"), $"S.classId" === $"C.classId") // joinType 默认 inner内连接
.show()
// 使用左外连接将df和frmClass根据classId合并
frmStu
.join(frmClass, Seq("classId"), "left")
.show()
// 左差集
frmStu
.join(frmClass, Seq("classId"), "anti")
.show()
// 左交集
frmStu
.join(frmClass, Seq("classId"), "semi")
.show()
결과
别名 + inner 内连接
+----+-------+-------+---------+
|name|classId|classId|className|
+----+-------+-------+---------+
|张三| 1| 1| name1|
|李四| 1| 1| name1|
|王五| 2| 2| name2|
|赵六| 2| 2| name2|
|李明| 2| 2| name2|
|王刚| 4| 4| name4|
+----+-------+-------+---------+
使用左外连接将df和frmClass根据classId合并
+-------+----+---------+
|classId|name|className|
+-------+----+---------+
| 1|张三| name1|
| 1|李四| name1|
| 2|王五| name2|
| 2|赵六| name2|
| 2|李明| name2|
| 4|王刚| name4|
| 5|王朋| null|
+-------+----+---------+
左差集
+-------+----+
|classId|name|
+-------+----+
| 5|王朋|
+-------+----+
左交集
+-------+----+
|classId|name|
+-------+----+
| 1|张三|
| 1|李四|
| 2|王五|
| 2|赵六|
| 2|李明|
| 4|王刚|
+-------+----+
그는 30년 이상 기술 연구에 전념해 왔으며, java, linux, javascript, php, css 등 다양한 언어에 능숙하며, 오픈 소스 분야에 많은 공헌을 했습니다. 나중에 참조할 수 있도록 기술 개발의 일부 문제를 공유하는 개발자 문서 스테이션입니다.
우편메소피아@프로톤메일.com