column对象：
| 分类 | 操作 | 解释 |
|---|---|---|
|
创建 |
|
单引号 |
|
|
同理, |
|
|
|
|
|
|
|
|
|
|
|
前面的 |
|
|
|
可以通过
|
|
|
别名和转换 |
|
|
|
|
通过 |
|
|
添加列 |
|
通过 |
|
操作 |
|
通过 |
|
|
通过 |
|
|
|
在排序的时候, 可以通过 |
// 1. 创建 spark 对象 val spark = SparkSession.builder() .master("local[6]") .appName("column") .getOrCreate() import spark.implicits._ @Test def creation(): Unit = { val ds: Dataset[Person] = Seq(Person("zhangsan", 15), Person("lisi", 10)).toDS() val ds1: Dataset[Person] = Seq(Person("zhangsan", 15), Person("lisi", 10)).toDS() val df: DataFrame = Seq(("zhangsan", 15), ("lisi", 10)).toDF("name", "age") // 2. ' 必须导入spark的隐式转换才能使用 str.intern() val column: Symbol = 'name // 3. $ 必须导入spark的隐式转换才能使用 val column1: ColumnName = $"name" // 4. col 必须导入 functions import org.apache.spark.sql.functions._ val column2: sql.Column = col("name") // 5. column 必须导入 functions val column3: sql.Column = column("name") // 这四种创建方式, 有关联的 Dataset 吗? ds.select(column).show() // Dataset 可以, DataFrame 可以使用 Column 对象选中行吗? df.select(column).show() // select 方法可以使用 column 对象来选中某个列, 那么其他的算子行吗? df.where(column === "zhangsan").show() // column 有几个创建方式, 四种 // column 对象可以用作于 Dataset 和 DataFrame 中 // column 可以和命令式的弱类型的 API 配合使用 select where // 6. dataset.col // 使用 dataset 来获取 column 对象, 会和某个 Dataset 进行绑定, 在逻辑计划中, 就会有不同的表现 val column4: sql.Column = ds.col("name") val column5: sql.Column = ds1.col("name") // 这会报错 // ds.select(column5).show() // 为什么要和 dataset 来绑定呢? // ds.join(ds1, ds.col("name") === ds1.col("name")) // 7. dataset.apply val column6: sql.Column = ds.apply("name") val column7: sql.Column = ds("name") }
@Test def as(): Unit = { val ds: Dataset[Person] = Seq(Person("zhangsan", 15), Person("lisi", 10)).toDS() // select name, count(age) as age from table group by name ds.select('name as "new_name").show() ds.select('age.as[Long]).show() }
@Test def api(): Unit = { val ds: Dataset[Person] = Seq(Person("zhangsan", 15), Person("lisi", 10)).toDS() // 需求一, ds 增加列, 双倍年龄 // 'age * 2 其实本质上就是将一个表达式(逻辑计划表达式) 附着到 column 对象上 // 表达式在执行的时候对应每一条数据进行操作 ds.withColumn("doubled", 'age * 2).show() // 需求二, 模糊查询 // select * from table where name like zhang% ds.where('name like "zhang%").show() // 需求三, 排序, 正反序 ds.sort('age asc).show() // 需求四, 枚举判断 ds.where('name isin ("zhangsan", "wangwu", "zhaoliu")).show() }
缺省值处理：
缺失值的处理思路
如果想探究如何处理无效值, 首先要知道无效值从哪来, 从而分析可能产生的无效值有哪些类型, 在分别去看如何处理无效值
什么是缺失值
-
一个值本身的含义是这个值不存在则称之为缺失值, 也就是说这个值本身代表着缺失, 或者这个值本身无意义, 比如说
null, 比如说空字符串
关于数据的分析其实就是统计分析的概念, 如果这样的话, 当数据集中存在缺失值, 则无法进行统计和分析, 对很多操作都有影响
缺失值如何产生的
-
Spark 大多时候处理的数据来自于业务系统中, 业务系统中可能会因为各种原因, 产生一些异常的数据
例如说因为前后端的判断失误, 提交了一些非法参数. 再例如说因为业务系统修改
MySQL表结构产生的一些空值数据等. 总之在业务系统中出现缺失值其实是非常常见的一件事, 所以大数据系统就一定要考虑这件事.
缺失值的类型
-
常见的缺失值有两种
-
null,NaN等特殊类型的值, 某些语言中null可以理解是一个对象, 但是代表没有对象,NaN是一个数字, 可以代表不是数字针对这一类的缺失值,
Spark提供了一个名为DataFrameNaFunctions特殊类型来操作和处理 -
"Null","NA"," "等解析为字符串的类型, 但是其实并不是常规字符串数据针对这类字符串, 需要对数据集进行采样, 观察异常数据, 总结经验, 各个击破
DataFrameNaFunctions
当数据集中出现缺失值的时候, 大致有两种处理方式, 一个是丢弃, 一个是替换为某值,
DataFrameNaFunctions中包含一系列针对空值数据的方案-
DataFrameNaFunctions.drop可以在当某行中包含null或NaN的时候丢弃此行 -
DataFrameNaFunctions.fill可以在将null和NaN充为其它值 -
DataFrameNaFunctions.replace可以把null或NaN替换为其它值, 但是和fill略有一些不同, 这个方法针对值来进行替换 -
// 1. 创建 SparkSession val spark = SparkSession.builder() .master("local[6]") .appName("null processor") .getOrCreate() @Test def nullAndNaN(): Unit = { // 2. 导入数据集 // 3. 读取数据集 // 1. 通过Saprk-csv自动的推断类型来读取, 推断数字的时候会将 NaN 推断为 字符串 // spark.read // .option("header", true) // .option("inferSchema", true)//这里是推断 // .csv(...) // 2. 直接读取字符串, 在后续的操作中使用 map 算子转类型 // spark.read.csv().map( row => row... ) // 3. 指定 Schema, 不要自动推断 val schema = StructType( List( StructField("id", LongType), StructField("year", IntegerType), StructField("month", IntegerType), StructField("day", IntegerType), StructField("hour", IntegerType), StructField("season", IntegerType), StructField("pm", DoubleType) ) ) val sourceDF = spark.read .option("header", value = true) .schema(schema) .csv("dataset/beijingpm_with_nan.csv") sourceDF.show() // 4. 丢弃 // 2019, 12, 12, NaN // 规则: // 1. any, 只有有一个 NaN 就丢弃 sourceDF.na.drop("any").show() sourceDF.na.drop().show() // 2. all, 所有数据都是 NaN 的行才丢弃 sourceDF.na.drop("all").show() // 3. 某些列的规则 sourceDF.na.drop("any", List("year", "month", "day", "hour")).show() // 5. 填充 // 规则: // 1. 针对所有列数据进行默认值填充 sourceDF.na.fill(0).show() // 2. 针对特定列填充 sourceDF.na.fill(0, List("year", "month")).show() }
遇到字符串的时候：
@Test def strProcessor(): Unit = { // 读取数据集 val sourceDF = spark.read .option("header", value = true) .option("inferSchema", value = true) .csv("dataset/BeijingPM20100101_20151231.csv") // sourceDF.show() // 1. 丢弃 import spark.implicits._ // sourceDF.where('PM_Dongsi =!= "NA").show() // 2. 替换 import org.apache.spark.sql.functions._ // select name, age, case // when ... then ... // when ... then ... // else sourceDF.select( 'No as "id", 'year, 'month, 'day, 'hour, 'season, when('PM_Dongsi === "NA", Double.NaN)//当啥的时候进行替换 .otherwise('PM_Dongsi cast DoubleType)//转换类型 .as("pm") ).show() // 原类型和转换过后的类型, 必须一致 sourceDF.na.replace("PM_Dongsi", Map("NA" -> "NaN", "NULL" -> "null")).show() //指定哪一行进行怎样的替换 }
-
-