제플린 걸음마 서울시립대학교 데이터마이닝 활용사례 제플린 노트북 통계 추출 코드

PrintZeppelinNotebookLanguageStatictics.sql

%sql
SELECT  COALESCE(lang, "Total") AS language, COUNT(*) AS cnt
FROM    lang
GROUP BY lang WITH ROLLUP
ORDER BY cnt DESC

ZeppelinNotebookLanguageStatictics.scala

import org.apache.spark.sql.functions._

// Change ZEPPELIN_NOTEBOOK_PATH!
val jsonRdd = sc.wholeTextFiles("<ZEPPELIN_NOTEBOOK_PATH>/*/note.json").map(_._2)
// For Spark 1.X
// val notes = sqlContext.read.json(jsonRdd)
val notes = spark.read.json(jsonRdd)

// Filter tutorial notes
val myNotes = notes.filter(r => !r.getString(4).contains("Tutorial"))
println(s"#Notes = ${myNotes.count}")

// Create paragraph df
val paragraphs = myNotes.selectExpr("EXPLODE(paragraphs)")
    .filter("col.text IS NOT NULL")
println(s"#Paragraphs = ${paragraphs.count}")

// UDF which extracts language from paragraph
val getLang = udf { s: String => s match {
        case _ if s.trim.length < 1 => "empty"
        case _ if s.startsWith("%") => 
            "%\\s?[\\w]+".r.findFirstIn(s) match {
                case Some(lang) => lang.replaceAll(" ", "")
                case None => "unknown"
            }
        case _ => "%spark"
    }
}

// For Spark 1.X
// paragraphs.withColumn("lang", getLang($"col.text"))
//    .registerTempTable("lang")
paragraphs.withColumn("lang", getLang($"col.text"))
    .createOrReplaceTempView("lang")