<jupyter_start><jupyter_text>read data and transform in traditional way<jupyter_code>sc
val rawblocks = sc.textFile("D:\\learn_spark\\donation") // load file
rawblocks.first
val head = rawblocks.take(10)
head.foreach(println)
def isHeader(line: String) = line.contains("id_1")
head.filter(x => !isHeader(x)).length  //一种调用函数的写法
val noheader = rawblocks.filter(!isHeader(_))  //另一种匿名函数的写法
noheader.first<jupyter_output><empty_output><jupyter_text>Data Frame<jupyter_code>val prev = spark.read.csv("D:\\learn_spark\\donation")
prev.show()
val parsed = spark.read.
    option("header", "true").
    option("nullValue", "?").
    option("inferSchema", "true").
    csv("D:\\learn_spark\\donation")
parsed.printSchema()
parsed.show()
parsed.cache()
parsed.count()
parsed.
    groupBy("is_match").
    count().
    orderBy($"count".desc).
    show()<jupyter_output>+--------+-------+
|is_match|  count|
+--------+-------+
|   false|5728201|
|    true|  20931|
+--------+-------+<jupyter_text>Spark SQL<jupyter_code>parsed.createOrReplaceTempView("linkage")  //定义关联名称为linkage
spark.sql("""
    SELECT is_match, COUNT(*) cnt
    FROM linkage
    GROUP BY is_match
    ORDER BY cnt DESC
""").show()
val summary = parsed.describe()
summary.show()  //fast summary
summary.select("summary", "cmp_fname_c1", "cmp_fname_c2").show()
val matches = parsed.where("is_match = true")
val matchSummary = matches.describe()

val misses = parsed.filter($"is_match" === false)
val missSummary = misses.describe()
misses.show()
summary.select("summary", "cmp_fname_c1", "cmp_fname_c2").show()
val schema = summary.schema
val longForm = summary.flatMap(row => {
    val metric = row.getString(0)
    (1 until row.size).map(i => {
        (metric, schema(i).name, row.getString(i).toDouble)
    })
})
val rdd = sc.parallelize(List("coffee panda","happy panda","happiest panda party"))
rdd.map(x=>x).collect
rdd.flatMap(x=>x.split(" ")).collect
val test = summary.flatMap(row => {
    val metric = row.getString(0)
    (1 until 2).map(i => {
        (metric, i)  //每一行映射成新的若干行
    })
})
test.show()
val longDF = longForm.toDF("metric", "field", "value")
longDF.show()
val wideDF = longDF.
    groupBy("field").
    pivot("metric", Seq("count", "mean", "stddev", "min", "max")).
    agg(first("value"))
//unknown error
:load Pivot.scala  
// transpose function

import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.functions.first

def pivotSummary(desc: DataFrame): DataFrame = {
    val schema = desc.schema
    import desc.sparkSession.implicits._

    val lf = desc.flatMap(row => {
        val metric = row.getString(0)
        (1 until row.size).map(i => {
            (metric, schema(i).name, row.getString(i).toDouble)
        })
    }).toDF("metric", "field", "value")

    lf.groupBy("field").
    pivot("metric", Seq("count", "mean", "stddev", "min", "max")).
    agg(first("value"))
}
val matchSummaryT = pivotSummary(matchSummary)
val missSummaryT = pivotSummary(missSummary)<jupyter_output><empty_output><jupyter_text>Joining DataFrames and Selecting Features<jupyter_code>matchSummaryT.createOrReplaceTempView("match_desc")
missSummaryT.createOrReplaceTempView("miss_desc")

spark.sql("""
    SELECT a.field, a.count+b.count total, a.mean - b.mean delta
    FROM match_desc a INNER JOIN miss_desc b ON a.field = b.field
    WHERE a.field NOT IN ("id_1", "id_2")
    ORDER BY delta DESC, total DESC
""").show()<jupyter_output>+------------+---------+--------------------+
|       field|    total|               delta|
+------------+---------+--------------------+
|     cmp_plz|5736289.0|  0.9563812499852176|
|cmp_lname_c2|   2464.0|  0.8064147192926264|
|      cmp_by|5748337.0|  0.7762059675300512|
|      cmp_bd|5748337.0|   0.775442311783404|
|cmp_lname_c1|5749132.0|  0.6838772482590526|
|      cmp_bm|5748337.0|  0.5109496938298685|
|cmp_fname_c1|5748125.0|  0.2854529057460786|
|cmp_fname_c2| 103698.0| 0.09104268062280008|
|     cmp_sex|5749132.0|0.032408185250332844|
+------------+---------+--------------------+
{'fraction_non_alphanumeric': 0.20044764983834867, 'fraction_numerical': 0.06341706043272818, 'mean_word_length': 3.4888392857142856, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 24, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '4', 'dir': 'jupyter-structured-clean-dedup', 'id': '128873', 'license': 'MIT', 'n_tokens_mistral': 1637, 'n_tokens_neox': 1484, 'n_words': 262, 'path': 'advanced_analytics_with_spark/chapter02_Introduction.ipynb', 'repo_name': 'XC-Li/Hadoop_Spark_Practice', 'size': '4021'}