校园大数据—数据清洗

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.api.java.UDF1;
import org.apache.spark.sql.types.DataTypes;
public class CleanData {
    public static void main(String[] args) {
        /********** Begin **********/
        SparkSession spark = SparkSession.builder().master("local").appName("clean").getOrCreate();
        Dataset<Row> data = spark.read().option("header", "true").csv("/data/workspace/myshixun/clean/cleandata.txt");
        // 去掉空值
        Dataset<Row> data1 = data.na().drop();
        // 身份证脱敏
        spark.udf().register("idEncrypt", (UDF1<String, String>) idcard ->
             idcard.replaceAll("(?<=\\w{2})\\w(?=\\w{2})", "*")
        , DataTypes.StringType);
        // 手机脱敏
        spark.udf().register("mobileEncrypt", (UDF1<String, String>) phone ->
             phone.replaceAll("(\\d{3})\\d{4}(\\d{4})", "$1****$2")
        , DataTypes.StringType);
        // 时间数据格式化
        spark.udf().register("format", (UDF1<String, String>) da -> da.substring(0, 4) + "-" + da.substring(4, 6) + "-" + da.substring(6), DataTypes.StringType);
        data1.registerTempTable("data");
//        spark.sql("select id,name,format(birth)birth,sex,address,idEncrypt(idcard)idcard,mobileEncrypt(phone)phone,email from data").show();
        spark.sql("select id,name,format(birth)birth,sex,address,idEncrypt(idcard)idcard,mobileEncrypt(phone)phone,email from data").write().option("header", "true").csv("/root/files");
        spark.stop();
        /********** End **********/
    }
}