<jupyter_start><jupyter_text>Text ProcessingAll tasks are words counting. It is find the main challenge is pre-processing the raw data into normal and clean form.1. Spark Session:sparkSession will give a context as well called "sparkSession.sparkContext". The default sparkSession is "spark", and sparkContext is "sc".2. Read data:txt file:```val textRDD = sparkSession.sparkContext.textFile("files/TaleOfTwoCities.txt")val textDS = sparkSession.read.textFile("C:/Users/Matthew/Desktop/TaleOfTwoCities.txt")val textDF = sparkSession.read.text("C:/Users/Matthew/Desktop/TaleOfTwoCities.txt")val raw1RDD = sparkSession.sparkContext.textFile("files/docword.enron.txt.gz") // only one partition```csv file:```val raw1RDD = sparkSession.sparkContext.textFile("files/CensusIncomeData.csv", 1)val raw1DF = sparkSession.read.    format("csv").    option("header", "false").    load("files/Data.csv") // only one partition```3. Access into first element of data and check length:List or Seq: a(0), a.lengthArray: a(0), a.lengthTuple: a.\_1Row: a.get(0) or a(0) (Type Any)    a.getDouble(0) (Type Double)    a(0).asInstanceOf\[Double\] (Type Double)    a.length4. Check number of partitions:```dataRDD.partitions.sizedataDF.rdd.partitions.size```<jupyter_code>import org.apache.spark.sql.SparkSession
import org.apache.spark.ml.{Pipeline, PipelineModel}
import org.apache.spark.ml.feature.{Tokenizer, StopWordsRemover}
import org.apache.spark.sql.Row
val sparkSession = SparkSession.builder.
    master("local[4]").
    appName("Text Processing").
    getOrCreate()<jupyter_output><empty_output><jupyter_text>Task 1: pipeline1. Read txt file into RDD.2. Use "pipeline" to change the capital words into small, and remove the stop words.3. Use "explode" to remove the empty row in DataFrame.4. Use "withColumn" to append a column, use "agg" to append a column with "groupBy"<jupyter_code>// Read data
val rawRDD = sparkSession.sparkContext.textFile("files/AnimalFarmChap1.txt", 20)
val wordsDF = rawRDD.flatMap(_.split("\\W").filter(_!="")).toDF("words")

// Build pipeline
val tokenizer = new Tokenizer().
    setInputCol("words").
    setOutputCol("smallWords")

val remover = new StopWordsRemover().
    setInputCol("smallWords").
    setOutputCol("smallNoStopWords")

val pipeline = new Pipeline().
    setStages(Array(tokenizer, remover))

val model = pipeline.fit(wordsDF)
val pipedDF = model.transform(wordsDF)

// explode used for flatten data. But here, we used it to remove empty cell
val dataDF = pipedDF.withColumn("explodedWords", explode($"smallNoStopWords")) // "withColumn" is used to add column, so do agg. But "agg" used with "groupBy"
val mostFreqDF = dataDF.groupBy("explodedWords").agg(count("*") as "TopFreqWords").orderBy(desc("TopFreqWords")) // "count" can automatically add a column named by count

// Count words which is at least 6 characters
val words6DF = rawRDD.flatMap(_.split("\\W").filter(_!="").filter(_.length >= 6)).toDF("words")
val piped6DF = model.transform(words6DF) // model do not need to retrain, as there is only transformer in pipeline
val data6DF = piped6DF.withColumn("explodedWords", explode($"smallNoStopWords"))
val mostFreq6DF = data6DF.groupBy("explodedWords").agg(count("*") as "TopFreqWords").orderBy("TopFreqWords")<jupyter_output><empty_output><jupyter_text>Task 2:1. Read csv file into RDD2. Remove the last incomplete element.<jupyter_code>val raw1RDD = sparkSession.sparkContext.textFile("files/CensusIncomeData.csv", 1)
val fileEnd = raw1RDD.count.toInt // convert long into int for arithmetic operation
val rawRDD = raw1RDD.mapPartitions(_.take(fileEnd-1)).repartition(20) // remove the last empty line.
val dataRDD = rawRDD.map(_.split(",").map(_.trim))
val dataDF = dataRDD.map(_(3)).toDF("degree")
dataDF.groupBy("degree").count.show(1)<jupyter_output>+-------+-----+
| degree|count|
+-------+-----+
|Masters| 1341|
+-------+-----+
only showing top 1 row<jupyter_text>Task 3:1. Read csv into DataFrame directly. Only 1 partition.2. Get the element in Row.<jupyter_code>// Read data direct into DataFrame. This way avoid error "ArrayIndexOutOfBoundsException" caused by the last row do not contain "degree".
// The "degree" in the last line is filled with "null".
val raw1DF = sparkSession.read.format("csv").
    option("header", "false").
    load("files/CensusIncomeData.csv")
val dataDF = raw1DF.select("_c3").repartition(20)
val sudDataDF = dataDF.filter(x => (x.get(0)==" Masters") || (x.get(0)==" 10th")) // x is Row, use "get" to obtaint the element in Row.
val countDF = sudDataDF.groupBy("_c3").count
countDF.show<jupyter_output>+--------+-----+
|     _c3|count|
+--------+-----+
|    10th|  753|
| Masters| 1341|
+--------+-----+<jupyter_text>Task 4:1. Read gz file. Only one partition.2. Convert RDD into tuple type, and the number of elements in tuple will be the number of DF columns.<jupyter_code>val raw1RDD = sparkSession.sparkContext.textFile("files/docword.enron.txt.gz", 1)
val rawRDD = raw1RDD.mapPartitions(_.drop(3)).repartition(20)
val dataRDD = rawRDD.map(_.split("\\W")).map(x => (x(1), x(2))) // to converted into two columns DataFrame, we need tuple whose first element is accessed by x._1
val dataDF = dataRDD.toDF("wordID", "count")
dataDF.groupBy("wordID").count.show<jupyter_output>+------+-----+
|wordID|count|
+------+-----+
| 18130|  795|
| 14204|  802|
| 26005|   71|
| 23843| 1505|
| 17686| 2619|
| 23097|   52|
| 12394|  138|
| 11888|  422|
| 24269|  494|
| 14369|   23|
| 20569|  607|
| 21452|   83|
| 14157|  229|
| 25555|  645|
| 14887|  877|
| 23459|   77|
|  1159|  430|
| 25032|  250|
|   467|  174|
| 18726|  138|
+------+-----+
only showing top 20 rows
{'fraction_non_alphanumeric': 0.16070796460176992, 'fraction_numerical': 0.04283185840707965, 'mean_word_length': 4.556538839724681, 'pattern_counts': {'":': 0, '<': 18, '<?xml version=': 0, '>': 21, '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': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '650047', 'license': 'Apache-2.0', 'n_tokens_mistral': 2111, 'n_tokens_neox': 1912, 'n_words': 509, 'path': 'TextProcessing.ipynb', 'repo_name': 'MatthewSZhang/SparkNotes', 'size': '5650'}