SparkSQL和Spark常用语句

Spark SQL 常用语句
读取数据
scala
val df = spark.read.option(“header”, “true”).csv(“path/to/csvfile.csv”)
val df = spark.read.parquet(“path/to/parquetfile.parquet”)
val df = spark.read.json(“path/to/jsonfile.json”)

展示数据
scala
df.show()
df.show(5) // 展示前5条记录

打印模式（Schema）
scala
df.printSchema()

选择列
scala
df.select(“column1”, “column2”).show()
df.select($“column1”, $“column2” + 1).show()

过滤数据
scala
df.filter($“column1” > 100).show()
df.filter(“column1 > 100”).show()

分组和聚合
scala
df.groupBy(“column1”).avg(“column2”).show()
df.groupBy(“column1”).agg(Max(“column2”), Sum(“column3”)).show()

排序
scala
df.orderBy($“column1”.desc).show()
df.orderBy(“column1”, “column2”.asc).show()

重命名列
scala
df.withColumnRenamed(“oldName”, “newName”).show()

添加新列
scala
df.withColumn(“newColumn”, $“column1” + $“column2”).show()

删除列
scala
df.drop(“column1”).show()

SQL 查询
scala
df.createOrReplaceTempView(“table_name”)
spark.sql(“SELECT * FROM table_name WHERE column1 > 100”).show()

连接（Join）
scala
val df1 = …
val df2 = …
df1.join(df2, df1(“id”) === df2(“id”)).show()
df1.join(df2, Seq(“id”)).show() // 适用于相同列名的简单连接

写入数据
scala
df.write.option(“header”, “true”).csv(“path/to/output.csv”)
df.write.parquet(“path/to/output.parquet”)

Spark 常用语句（Spark Core）
初始化 SparkConf 和 SparkContext
scala
val conf = new SparkConf().setAppName(“AppName”).setMaster(“local[*]”)
val sc = new SparkContext(conf)

读取数据
scala
val data = sc.textFile(“path/to/textfile.txt”)

展示数据
scala
data.take(10).foreach(println)

映射（Map）
scala
val mappedData = data.map(line => line.split(",")(0))

过滤（Filter）
scala
val filteredData = data.filter(line => line.contains(“keyword”))

flatMap
scala
val flatMappedData = data.flatMap(line => line.split(" "))

行动操作（例如：collect, count, reduce）
scala
data.collect()
data.count()
val reducedData = data.reduce(_ + _)

键值对操作
scala
val pairs = data.map(line => (line.split(",")(0), line.split(",")(1)))
val groupedByKey = pairs.groupByKey()
val reducedByKey = pairs.reduceByKey(_ + _)

排序
scala
val sortedData = data.sortBy(line => line.length)

联合（Union）
scala
val rdd1 = …
val rdd2 = …
val unionedRDD = rdd1.union(rdd2)

缓存（Cache）
scala
data.cache()

保存数据
scala
data.saveAsTextFile(“path/to/output”)