建表语句
首先是建表语句，百度热搜一共有热搜，小说，电影等六个主分类，消息种类中保存的就是这些种类的英文名称，也是url的参数，比如其为热搜的时候，对应的地址为https://top.baidu.com/board?tab=realtime，而如果变成小说则会变成https://top.baidu.com/board?tab=novel，tab后面的参数值就代表了其分类的不同。
/
createtable if not exists hot_context( id int primary key auto_increment comment '自增主键', key_kind varchar(255) comment '消息种类', time_send long comment '消息获取时间', hot_num int comment '消息热度', value_context text comment '消息缩写', value_all text comment '消息内容') comment ='百度热搜消息表';
然后消息热度表示爬取消息时，消息的热搜指数；消息缩写和消息内容则是消息的缩略以及简述，不过有部分消息是没有简述的，这里会置为null，下图中邓超情人节……和金莎……其实都是value_context中的内容，14日，孙俪……则为value_all的内容。
导入依赖
importurllib.requestasrequestfromlxmlimportetreeimporttimeimportcalendarfrompyspark.sqlimportSparkSessionfrompyspark.sqlimportRowimportrandomimportlogging
urllib是python中爬取网站的库，其有四个主要模块，这里只使用到了request[打开和读取url]的功能。
然后是lxml，这里使用xpath来解析爬取出来的html数据，写法和正则类似，比如我想爬取div块中class为a的里面的文本数据，可以写为etree.HTML(html数据).xpath('//div[@class,"a"]/text()')，其后的time和calendar则是用来获得当前消息时间戳的。
sparksession用来创建spark会话。
row则是用来封装数据，将爬取到的数据变成row格式，再写入dataframe中。
random是用来从几个UserAgent中选择其中一个，当然如果手头有多个proxies，也可以使用random.choice来选择其中一个。
logging用来打印日志信息。
尝试连接百度热搜
# 尝试连接百度热搜网站defgetBaiduHtml(opener, req_url): MAX_RETRY=6fortriesinrange(MAX_RETRY): try: # 尝试10sweb_txt=opener.open(req_url, timeout=10).read().decode('utf8') returnweb_txtexceptExceptionase: logging.warning(str(e) +',html_url:{0}'.format(req_url)) time.sleep(3) iftries< (MAX_RETRY-1): continueelse: print('Has tried {0} times to access url {1}, all failed!'.format(MAX_RETRY, req_url)) returnNone
这一段代码基本是从网上copy的，就是在链接超时后再次尝试，一直尝试够六次，当时在链接的时候，最初采用的代理是http格式的，但是百度热搜是https，所以代理失效，走的本地，后来增加了https的代理，却一直无法使用，以为是网络超时，结果是代理的问题，这段代码就保留了下来没有删除，此处返回的数据为html的源文件数据。
解析百度html数据
defgetBaiduData(): list_res= [] # 百度热搜前缀url_text="https://top.baidu.com/board?tab="# 代理UserAgent=random.choice( ['Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/41.0.2228.0 Safari/537.36', 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_10_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/41.0.2227.1 Safari/537.36', 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/41.0.2227.0 Safari/537.36']) headers= {'User-Agent': UserAgent} # 如果有代理，此处可以设置代理 {'http':'','https':''},为空，则使用的是本地ipproxies= {} # 百度热搜数组hot_arr= ["realtime", "novel", "movie", "teleplay", "car", "game"] forhot_textinhot_arr: req_url=request.Request(url=url_text+hot_text, headers=headers) # 获得handler对象handler=request.ProxyHandler(proxies=proxies) # 获得opener对象opener=request.build_opener(handler) web_txt=getBaiduHtml(opener, req_url) web_html=etree.HTML(web_txt).xpath('//div[contains(@class,"c-single-text-ellipsis") or ''contains(@class,"hot-desc_1m_jR small_Uvkd3 ") or''contains(@class,"hot-index_1Bl1a")]/text()') # 中间数据tmp_res=""forhtml_txtinweb_html: new_html_txt=html_txt.strip() ifnew_html_txt!=""andnew_html_txt.isdigit(): iftmp_res!="": list_res.append(Row(tmp_res)) tmp_res=hot_text+"\t"+str(calendar.timegm(time.gmtime())) +"\t"+new_html_txtelifnew_html_txt!=""andnotnew_html_txt.isdigit(): tmp_res=tmp_res+"\t"+new_html_txtelse: passreturnlist_res
代码中的百度热搜数组对应的就是上面提到的六个热搜类别，这里没有直接采用urlopen来获得数据，不是urlopen无法获得数据，只是想增加一些小难度，先用request.Request返回请求对象，然后用reques.ProxyHandler来获得handler对象，之后获得opener对象，大致说法如下：
- 使用相关的 Handler处理器 来创建特定功能的处理器对象；
- 然后通过 urllib.request.build_opener()方法使用这些处理器对象，创建自定义opener对象；
- 使用自定义的opener对象，调用open()方法发送请求。
此步骤是因为基本的urlopen()方法不支持代理、cookie等其他的HTTP/HTTPS高级功能。
然后根据xpath解析出全文三个不同内容的class，其对应可以在网站自行查看，大致是热搜指数，缩略和简述。
下面的判断逻辑是依据是爬取解析出来的第一个数据是热搜指数，也就是数字，当为数字的时候会将当前热搜分类，时间戳和指数变成以tab键分割的字符串，而不是数字的时候，则会将其他内容追加到后面，当再次出现数字的时候，会将该字符串变成row并写入到一个list集合中，字符串又变成当前热搜分类，时间戳和指数变成以tab键分割的字符串。
主方法
这里没有将写入mysql和df的转化单独抽离出来，一方面是因为难度不大，一方面是因为懒🦥。
if__name__=='__main__': spark=SparkSession.builder.appName("sql").master("local").getOrCreate() df=spark.createDataFrame(getBaiduData()) df.createTempView("tmpTable1") prop= { 'user': 'root', 'password': '', 'driver': 'com.mysql.cj.jdbc.Driver' } jdbc_url='jdbc:mysql://localhost:3306/baiduhot'# 格式化row中数据sql_format=spark.sql( "select split_part(_1,'\\t',1) as key_kind,""split_part(_1,'\\t',2) as time_send,""split_part(_1,'\\t',3) as hot_num,""split_part(_1,'\\t',4) as value_context,""case when length(split_part(_1,'\\t',5))>0 then split_part(_1,'\\t',5) end as value_all ""from tmpTable1" ) # 写入数据库sql_format.write.jdbc(url=jdbc_url, table='hot_context', mode='append', properties=prop) print("the success") spark.stop()
方法内均为spark代码，master设为local，只在idea单独运行，第三行创建的df是解析百度数据后返回的数据，第四行是将该df注册成临时表tmpTable1，prop中是连接mysql需要的信息，没有使用df本身的方法来解析数据，而是采用sql，道理依旧是简单。
最后就是正常用追加的方式将数据写入到表中，需要注意将jdbc的jar包放在spark的jars目录下，否则会报错。
