Spark使用Jieba分词Hive数据生成新表（五）

package lishijia.spark.demo.hivejieba

import com.huaban.analysis.jieba.{JiebaSegmenter, SegToken}
import com.huaban.analysis.jieba.JiebaSegmenter.SegMode
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.functions._

object HiveJieba {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf() .registerKryoClasses(Array(classOf[JiebaSegmenter]))
        .set("spark.rpc.message.maxSize", "800")

    val spark = SparkSession
      .builder()
      .appName("Jieba udf")
      .enableHiveSupport()
      .config(conf)
      .getOrCreate()

    val df =spark.sql("select sentence,label from lishijia.news_seg limit 300")
    // 调用分词方法
    val df_seg = jieba_seg(df,"sentence")
    df_seg.show()
    /**
      * DataFrame对象生成表
      */
    df_seg.write.mode("overwrite").saveAsTable("lishijia.news_jieba")
  }

  /**
    * 分词方法
    * @param df（DataFrame，需要处理数据源）
    * @param columnName（DataFrame数据源需要处理的列）
    * @return
    */
  def jieba_seg(df:DataFrame, columnName:String) : DataFrame = {
    val segmenter = new JiebaSegmenter()
    // 通过spark上下文把jieba分词对象发送到广播中（其他节点通过广播使用）
    val seg = df.sparkSession.sparkContext.broadcast(segmenter)
    // 定义udf函数
    val jieba_udf = udf{ (sentence:String) =>
        // 通过spark广播获取jieba分词对象(分布式)
        val segV = seg.value
        // jieba分词对象处理一个句子
        segV.process(sentence.toString, SegMode.INDEX)
          .toArray().map(_.asInstanceOf[SegToken].word)
          .filter(_.length>1).mkString("/")
    }
    /**
      * 调用udf分词函数
      */
    df.withColumn("seg", jieba_udf(col(columnName)))
  }

}