SparkSql中Dataset(DataFrame)操作

分手后的思念是犯贱 2023-02-15 10:27 14阅读 0赞

参考文章：[Spark SQL中的Encoder][Spark SQL_Encoder]

[SparkSQL DataFrame和Dataset基本操作][SparkSQL DataFrame_Dataset]

[【SparkSQL笔记】SparkSQL的Dataset操作大全（二）][SparkSQL_SparkSQL_Dataset]

# DataFrame操作 #

## 操作方式举例： ##

/*
    * 数据集: 
    *   张三,23
    *   李四,24
    *   王五,25
    *   赵六,26
    */
    val spark = SparkSession
          .builder()
          .appName(this.getClass.getSimpleName)
          .master(master = "local[*]")
          .getOrCreate()
    
        import spark.implicits._
        val df = spark.read.textFile("./data/clm")
          .map(_.split(","))
          .map(x => (x(0), x(1)))
          .toDF("name", "age")
          .cache()
    // ========================================================= //
    
    /**
    *  withColumn:
    *    通过添加列或替换具有相同名称的现有列来返回新的数据集
    *     column的表达式只能引用此数据集提供的属性。 添加引用其他数据集的列是错误的
    *  
    *  DataFrame其它表示方式: df("age")
    */
    df.withColumn("new_age", col = df("age") + 1).show()
    
    // 结果:
    +----+---+-------+
    |name|age|new_age|
    +----+---+-------+
    |张三| 23|   24.0|
    |李四| 24|   25.0|
    |王五| 25|   26.0|
    |赵六| 26|   27.0|
    +----+---+-------+

## spark dataframe新增一列的四种方法： ##

/*
    *  spark dataframe新增一列的四种方法:
    *    方法一：利用createDataFrame方法，新增列的过程包含在构建rdd和schema中
    *    方法二：利用withColumn方法，新增列的过程包含在udf函数中
    *    方法三：利用SQL代码，新增列的过程直接写入SQL代码中
    *    方法四：以上三种是增加一个有判断的列，如果想要增加一列唯一序号，可以使用monotonically_increasing_id
    **/
    //dataframe新增一列方法1，利用createDataFrame方法
    val trdd = input.select(targetColumns).rdd.map(x=>{
      if (x.get(0).toString().toDouble > critValueR || x.get(0).toString().toDouble < critValueL) 
        Row(x.get(0).toString().toDouble,"F")
      else Row(x.get(0).toString().toDouble,"T")      
      })      
    val schema = input.select(targetColumns).schema.add("flag", StringType, true)
    val sample3 = ss.createDataFrame(trdd, schema).distinct().withColumnRenamed(targetColumns, "idx")
    
    //dataframe新增一列方法2
    val code :(Int => String) = (arg: Int) => {if (arg > critValueR || arg < critValueL) "F" else "T"}
    val addCol = udf(code)
    val sample3 = input.select(targetColumns).withColumn("flag", addCol(input(targetColumns)))
    .withColumnRenamed(targetColumns, "idx")
    
    //dataframe新增一列方法3
    input.select(targetColumns).createOrReplaceTempView("tmp")
    // sql示例: select user_id, 'ruleTagId' as rule_tag_id from tmp_view
    val sample3 = ss.sqlContext.sql("select distinct "+targetColname+
        " as idx,case when "+targetColname+">"+critValueR+" then 'F'"+
        " when "+targetColname+"<"+critValueL+" then 'F' else 'T' end as flag from tmp")
    
    //添加序号列
    import org.apache.spark.sql.functions.monotonically_increasing_id
    val inputnew = input.withColumn("idx", monotonically_increasing_id)

# #

### [Spark 中 关于scala 语法理解的问题][Spark _ _scala] ###

// 语法定义:
    def flatMap[U : Encoder](func: T => TraversableOnce[U]): Dataset[U] =
            mapPartitions(_.flatMap(func))
    // 调用示例:
    val lines: Dataset[String] = session.read.textFile("")
    val words: Dataset[String] = lines.flatMap(_.split(" "))
    
    // 说明:
    // _.split(" ") 相当于一个函数, 传入参数是String类型, 
    // 返回类型是split的结果类型 Array[String],
    
    // 而flatmap方法定义为接收 func: T => TraversableOnce[U], 
    // 显然这里的T是String类型,
    
    // 而 Array[String] 并不是 TraversableOnce[U] 类型, 
    // 因为 Array 没有实现 TraversableOnce 这个特质,
    
    // 传入 _.split(" ") 函数给 flatMap, flatMap为什么没有报错?
    // 原因: scala通过隐式转换默认调用了iterator 方法
    // 转换为一个可迭代的对象（Array(1,2,3).iterator）

==============================================================

以下错误，想必在做Spark的DateSet操作时一定是见过吧？

Error:(58, 17) Unable to find encoder for type stored in a Dataset.  
    Primitive types (Int, String, etc) and Product types (case classes) 
    are supported by importing spark.implicits._  Support for serializing 
    other types will be added in future releases.
        peopleDF.map(row => row.get(0)).show()

这是因为在作map转换时需要指定一个转换的Encorder,   
在Scala代码中是通过隐式转换进行的, 而在Java代码中则需要在代码中指明。  
  
Java代码如下：

public static void main(String[] args) {
        SparkSession spark = SparkSession.builder()
                            .master("local[2]")
                            .appName("DataSetEncoderExample")
                            .getOrCreate();
    
        List<String> data = Arrays.asList("chen", "li", "huang");
        
        //创建DataSet的时候指明数据是String类型
        Dataset<String> ds = spark.createDataset(data, Encoders.STRING());
    
        /***操作一：**/
        // map操作：把string类型的变换成string类型
        // 此时MapFunction<String, String> 这两个地方都应该是String
        Dataset<String> dsString2String = 
    	    ds.map((MapFunction<String, String>) 
    		    v -> "Hi," + v, Encoders.STRING());
        dsString2String.show();
    
        /***操作二：**/
        // map操作：把string类型的变换成int类型
        // 注意此时MapFunction<String, Integer> 这两个地方的类型变化
        // 第一个类型String为原来的DataSet的类型, 第二个类型为输出的类型
        Dataset<Integer> dsString2Int = 
    	    ds.map(new MapFunction<String, Integer>(){
                @Override
                public Integer call(String value) throws Exception {
                    return value.length();
                }
            }, Encoders.INT());
        dsString2Int.show();
    
        /***操作三：**/
        // map操作：把string类型的变换成自定义的对象类型
        // 注意此时MapFunction<String, People> 这两个地方的类型变化
        // 第一个类型String为原来的DataSet的类型, 第二个类型People为输出的类型
        Encoder<People> peopleEncoder = Encoders.kryo(People.class);
        Dataset<People> dsString2Object = ds.map(
    	    new MapFunction<String, People>(){
                @Override
                public People call(String value) throws Exception {
                    return new People(value, value.length());
                }
            }, peopleEncoder);
        dsString2Object.show();
        dsString2Object
                .map((MapFunction<People, String>) 
    			     item -> item.getName(), Encoders.STRING())
                .show();
    
        /***操作四：**/
        // map操作：把string类型的变换成Row对象类型
        // 注意此时MapFunction<String, Row> 这两个地方的类型变化
        // 第一个类型String为原来的DataSet的类型, 第二个类型Row为输出的类型
        // Encoder<Row> rowEncoder = Encoders.kryo(Row.class);
        Encoder<Row> rowEncoder = Encoders.javaSerialization(Row.class);
        Dataset<Row> dsString2Row = ds
                    .map((MapFunction<String, Row>) value 
                        -> RowFactory.create(value, value.length()), rowEncoder);
        dsString2Row.show();
        dsString2Row
                .map((MapFunction<Row, String>) 
    			    item -> item.getString(0), Encoders.STRING())
                .show();
    
        spark.stop();
    }

**People类：**

public class People {
        String name;
        Integer age;
    
        public People( String name, Integer age){
            this.name = name;
            this.age = age;
        }
    
        public String getName() {
            return name;
        }
    
        public void setName(String name) {
            this.name = name;
        }
    
        public Integer getAge() {
            return age;
        }
    
        public void setAge(Integer age) {
            this.age = age;
        }
    }

[Spark SQL_Encoder]: https://www.jianshu.com/p/d3c35e18af44
[SparkSQL DataFrame_Dataset]: https://blog.csdn.net/qq_33713328/article/details/89682989
[SparkSQL_SparkSQL_Dataset]: https://blog.csdn.net/sdut406/article/details/103445486
[Spark _ _scala]: https://segmentfault.com/q/1010000014159248