说明

读取 ods_yuqing.t_weibo 表，分析出以下数据：

• 微博id
• 微博原内容（带有html）
• 微博文本内容（去掉html标签）
• 微博发布时间
• 作者id
• 作者姓名
• 作者性别
• 作者粉丝数量

并将以上数据，保存到 dwd 层的 yuqing_dwd.t_weibo 表

代码

import json
import findspark
import datetime

findspark.init()

from pyspark.sql import  SparkSession 

def handleRow(row):
    from bs4 import BeautifulSoup
    # 取值content
    content = row["content"]
    jsonObj = json.loads(content)
    mblog = jsonObj.get("mblog")

    weiboId = mblog.get("id","")
    weiboHTML = mblog.get("text","") # 微博正文，html格式
    weiboText = BeautifulSoup(weiboHTML,'html.parser').get_text() # 微博正文，文本格式
    createdAt = mblog.get("created_at","")

    datetimeObj=datetime.datetime.strptime(createdAt,'%a %b %d %H:%M:%S %z %Y')
    createdAt2=datetimeObj.strftime('%Y-%m-%d %H:%M:%S')

    user = mblog.get("user")
    userId = user.get("id","")
    screenName = user.get("screen_name","")
    gender = user.get("gender","")
    followerCount = user.get("followers_count","")


    return userId,screenName,gender,followerCount,weiboId,createdAt2,weiboHTML,weiboText

def exe():
    # 本地运行
    spark = SparkSession.builder.\
        appName("odw2dwd").\
        master("local[1]").\
        enableHiveSupport().\
        getOrCreate()

    # yarn运行
    # spark = SparkSession.builder.\
    #     appName("odw2dwd").\
    #     enableHiveSupport().\
    #     getOrCreate()

    spark.sparkContext.addPyFile("hdfs:///yuqing/packages/yuqing_ods_dwd.zip")

    df = spark.sql("select * from ods_yuqing.t_weibo")

    resRDD = df.rdd.map( handleRow )
    # resRDD.foreach(print)

    df = spark.createDataFrame(resRDD,["user_id","screen_name","gender","follower_count","weibo_id","created_at","html","content"])

    # df.show()
    # df.printSchema()
    df.createOrReplaceTempView("t_weibo_temp")
    spark.sql("insert overwrite table yuqing_dwd.t_weibo select *,substr(created_at,1,10) from t_weibo_temp")

    spark.stop()

if __name__ == "__main__":
    exe()

原文出处：/show_1IX3Gd42QkWj.html