？ import requests
？ from bs4 import BeautifulSoup
？ import time
？ # 请求网页
？ def page_request(url, ua):
？     response = requests.get(url=url, headers=ua)
？     html = response.content.decode('utf-8')
？     return html
？ # 解析网页
？ def page_parse(html):
？     soup = BeautifulSoup(html, 'lxml')
？     # 获取每部电影的排名
？     position = soup.select('#content > div > div.article > ol > li > div > div.pic > em')
？     # 获取豆瓣电影名称
？     name = soup.select('#content > div > div.article > ol > li > div > div.info > div.hd > a > span:nth-child(1)')
？     # 获取电影评分
？     rating = soup.select('#content > div > div.article > ol > li> div > div.info > div.bd > div > span.rating_num')
？     # 获取电影链接
？     href = soup.select('#content > div > div.article > ol > li > div > div.info > div.hd > a')
？     for i in range(len(name)):
？         print(position[i].get_text()+'\t'+name[i].get_text()+'\t' + rating[i].get_text() + '\t' + href[i].get('href'))
？ 
？ if __name__ == "__main__":
？     print('**************开始爬取豆瓣电影**************')
？     ua = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4421.5 Safari/537.36'}
？     # 豆瓣电影Top250每页有25部电影，start就是每页电影的开头
？     for startNum in range(0, 251, 25):
？         url = "https://movie.douban.com/top250?start=%d" % startNum
？         html = page_request(url=url, ua=ua)
？         page_parse(html=html)
？         print('**************爬取完成**************')

（二）存储影片详细信息

访问豆瓣电影Top250（https://movie.douban.com/top250?start=0），在问题1的基础上，获取每部电影的导演、编剧、主演、类型、上映时间、片长、评分人数以及剧情简介等信息，并将获取到的信息保存至本地文件中。

参考步骤及过程：

1) 在实验1创建的工程包webCrawler下的doubanCrawler文件夹下创建downloadTop250.py；
2) 参考代码


？ import requests
？ from bs4 import BeautifulSoup
？ import re
？ import docx
？ from docx.oxml.ns import qn
？ 
？ # 请求网页
？ def page_request(url, ua):
？     response = requests.get(url=url, headers=ua)
？     html = response.content.decode('utf-8')
？     return html
？ 
？ # 解析网页
？ def page_parse(html, ua):
？     soup = BeautifulSoup(html, 'lxml')
？     for tag in soup.find_all(attrs={'class': 'item'}):
？         data = []
？         # 序号
？         num = tag.find('em').get_text()
？         data.append(num)
？         # 电影名称
？         name = tag.find_all(attrs={'class': 'title'})[0].get_text()
？         data.append(name)
？         # 豆瓣链接
？         href = tag.find(attrs={'class': 'hd'}).a
？         url = href.attrs['href']
？         data.append('豆瓣链接:' + url)
？         # 评分与评论数
？         info = tag.find(attrs={'class': 'star'}).get_text()
？         info = info.replace('\n', '').lstrip()
？         # 使用正则表达式获取数字
？         mode = re.compile(r'\d+\.?\d')
？         i = 0
？         for n in mode.findall(info):
？             if i == 0:
？                 # 评分
？                 data.append('豆瓣评分:' + n)
？             elif i == 1:
？                 # 评分人数
？                 data.append('评分人数:' + n)
？             i = i + 1
？         # 进入子网页，获取每部电影的具体信息
？         sub_page_requests(url, ua, data)
？         print('第%s部电影信息爬取完成' % num)
？ 
？ # 子网页处理函数：进入并解析子网页/请求子网页
？ # 获取影片详细信息
？ def sub_page_requests(url, ua, data):
？     html = page_request(url=url, ua=ua)
？     soup = BeautifulSoup(html, 'lxml')
？     # 影片信息
？     info = soup.find(attrs={'id': 'info'}).get_text()
？     data.append(info)
？     # 影片简介
？     summary = soup.find(attrs={'property': 'v:summary'}).get_text()
？     summary = summary.replace('\n', '').replace(' ', '').lstrip()
？     data.append(data[1] + '影片简介:\n' + summary)
？     # 保存影片信息
？     save(data)
？ 
？ def save(data):
？     file = docx.Document()
？     # 设置字体格式
？     file.styles['Normal'].font.name = u'Times New Roman'
？     file.styles['Normal'].element.rPr.rFonts.set(qn('w:eastAsia'), u'Times New Roman')
？ 
？     # 将爬取到的数据写入word中
？     for element in data:
？         file.add_paragraph(element)
？     file.save('result/' + data[0] + '、' + data[1] + '.docx')
？ 
？ if __name__ == "__main__":
？     print('**************开始爬取豆瓣电影**************')
？     ua = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4421.5 Safari/537.36'}
？     # 豆瓣电影Top250每页有25部电影，start就是每页电影的开头
？     data_List = []
？     for startNum in range(0, 251, 25):
？         url = "https://movie.douban.com/top250?start=%d" % startNum
？         html = page_request(url=url, ua=ua)
？         # 获取每部影片的信息
？         page_parse(html=html, ua=ua)
？         print('**************爬取完成**************')

熟悉Flume的基本使用方法

2．Kafka链接Flume

编写Flume配置文件，将Kafka作为输入，通过Flume将Kafka生产者输入的信息存入HDFS中，存储格式hdfs://localhost:9000/fromkafka/%Y%m%d/，要求存储时文件名为kafka_log。(注：配置好Flume后生产者输入的信息不会实时写入到HDFS，一段时间后批量写入)

【参考答案】
Flume配置文件如下：

？ #文件名Ekafka2hdfs.conf
？ #设置名称
？ a1.sources=r1
？ a1.sinks=k1
？ a1.channels=c1
？ 
？ #配置Source
？ a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource
？ a1.sources.r1.batchSize = 500
？ a1.sources.r1.batchDurationMillis = 2000
？ a1.sources.r1.kafka.bootstrap.servers = localhost:9092
？ a1.sources.r1.kafka.topics = flume
？ 
？ #配置Sink
？ a1.sinks.k1.type = hdfs
？ a1.sinks.k1.hdfs.path = hdfs://localhost:9000/fromkafka/%Y%m%d/
？ a1.sinks.k1.hdfs.filePrefix = kafka_log
？ a1.sinks.k1.hdfs.maxOpenFiles=5000
？ a1.sinks.k1.hdfs.fileType = DataStream
？ a1.sinks.k1.hdfs.batchSize = 100
？ a1.sinks.k1.hdfs.writeFormat=Text
？ a1.sinks.k1.hdfs.rollInterval = 60
？ a1.sinks.k1.hdfs.rollSize = 102400
？ a1.sinks.k1.hdfs.rollCount = 100000
？ a1.sinks.k1.hdfs.round = true
？ a1.sinks.k1.hdfs.roundValue = 10
？ a1.sinks.k1.hdfs.roundUnit = minute
？ a1.sinks.k1.hdfs.useLocalTimeStamp = true
？ 
？ #配置channels
？ a1.channels.c1.type=memory
？ a1.channels.c1.keep-alive=120
？ a1.channels.c1.capacity=500000
？ a1.channels.c1.transactionCapacity=600
？ 
？ #绑定sink source到channels上
？ a1.sources.r1.channels=c1
？ a1.sinks.k1.channel=c1


？ 新建一个cmd窗口，输入如下命令启动zookeeper：
？ > cd C:\kafka_2.12-2.4.0
？ > .\bin\windows\zookeeper-server-start.bat .\config\zookeeper.Properties
？ 打开一个新的cmd窗口，输入如下命令启动Kafka：
？ > cd C:\kafka_2.12-2.4.0
？ > .\bin\windows\kafka-server-start.bat .\config\server.properties
？ 再打开一个新的cmd窗口创建Kafka的topic：
？ > cd C:\kafka_2.12-2.4.0
？ > .\bin\windows\kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic flume
？ 创建完成之后创建Kafka生产者：
？ > cd C:\kafka_2.12-2.4.0
？ > .\bin\windows\kafka-console-producer.bat --broker-list localhost:9092 --topic flume
？ 打开新的cmd窗口开启HDFS：
？ > cd C:\hadoop-3.1.3\sbin
？ > start-dfs.cmd
？ 这时会弹出namenode和datanode窗口，不要关闭
？ 在新的cmd窗口启动flume：
？ > cd C:\apache-flume-1.9.0-bin
？ > .\bin\flume-ng agent --conf .\conf  --conf-file .\conf\Ekafka2hdfs.conf --name a1 -property flume.root.logger=INFO,console
？ 在启动生产者的窗口中输入“HelloFlume”。
？ 然后，打开浏览器输入“http://localhost:9870/explorer.html#”，查看HDFS中的文件，结果应如下：

基本题二：使用Flume写入当前文件系统

假设有一个目录“C:/mylog/”，现在新建两个文本文件1.txt与2.txt，在1.txt中输入“Hello Flume”，在2.txt中输入“hello flume”，请使用Flume对目录“C:/mylog/”进行监控，当把文件1.txt与2.txt放入该目录时，Flume就会把文件内容写入到“C:/backup”目录下的文件中。（注：配置文件中Source的类型为spooldir，Sink的类型为file_roll，具体用法可以参考文档：https://flume.apache.org/releases/content/1.9.0/FlumeUserGuide.html）

【参考答案】
Flume配置文件如下：

？ #文件名Efile2file.conf
？ #设置名称
？ a1.sources=r1
？ a1.sinks=k1
？ a1.channels=c1
？ 
？ #配置Source
？ a1.sources.r1.type=spooldir
？ a1.sources.r1.spoolDir=C:/mylog/
？ 
？ #配置Sink
？ a1.sinks.k1.type = file_roll
？ a1.sinks.k1.channel = c1
？ a1.sinks.k1.sink.directory = C:/backup
？ 
？ #配置channels
？ a1.channels.c1.type=memory
？ a1.channels.c1.capacity=1000
？ a1.channels.c1.transactionCapacity=100
？ 
？ #绑定sink source到channels上
？ a1.sources.r1.channels=c1
？ a1.sinks.k1.channel=c1

？ 在C盘创建两个目录，即mulog以及backup，再按要求创建两个txt文件，即1.txt和2.txt。
？ 打开一个cmd窗口启动Flume，命令如下：
？ > cd C:\apache-flume-1.9.0-bin
？ > .\bin\flume-ng agent --conf .\conf --conf-file .\conf\Efile2file.conf --name a1 -property flume.root.logger=INFO,console
？ 将1.txt和2.txt拖入mylogs目录中，然后查看backup目录中的文件，可以看到如下结果：

pandas数据清洗初级实践

基本题一基础练习

（1）根据列表[“Python”,”C”,”Scala”,”Java”,”GO”,”Scala”,”SQL”,”PHP”,”Python”]创建一个变量名为language的Series；

？ >>> import pandas as pd
？ >>> import numpy as np
？ >>> from pandas import DataFrame, Series
？ >>> language = Series(["Python","C","Scala","Java","GO","Scala","SQL","PHP","Python"])

（2）创建一个由随机整型组成的Series，要求长度与language相同，变量名为score；

1	？ >>> score = Series(np.random.randint(10, size = (9,)))

（3）根据language和score创建一个DataFrame；

？ >>> df = DataFrame({'language':language, score':score})
？ >>> df
？ language  score 
？ 0   Python      8
？ 1        C      6
？ 2    Scala      8
？ 3     Java      4
？ 4       GO      9 
？ 5    Scala      8
？ 6      SQL      5
？ 7      PHP      1
？ 8   Python      1

（4）输出该DataFrame的前4行数据；

？ >>> df.head(4)
？ language  score   
？ 0   Python      8
？ 1        C      6 
？ 2    Scala      8 
？ 3     Java      4

（5）输出该DataFrame中language字段为Python的行；

？ >>> df[df.language == 'Python']
？ language  score
？ 0   Python      8
？ 8   Python      1

（6）将DataFrame按照score字段的值进行升序排序；

？ >>> df.sort_values('score',inplace=True)
？ >>> df
？ language  score
？ 7      PHP      1
？ 8   Python      1
？ 3     Java      4
？ 6      SQL      5
？ 1        C      6
？ 0   Python      8
？ 2    Scala      8
？ 5    Scala      8
？ 4       GO      9

（7）统计language字段中每种编程语言出现的次数。

？ >>> df.language.value_counts()                                                                                          Scala     2
？ Python    2
？ PHP       1
？ GO        1
？ SQL       1
？ C         1 
？ Java      1
？ Name: language, dtype: int64

2. 酒类消费数据

1) 用pandas将酒类消费数据表中的数据读取为DataFrame，输出包含缺失值的行；

？ >>> import pandas as pd
？ >>> df = pd.read_csv('D:\drinks.csv')
？ >>> df.loc[df.continent.isnull()].index
？ Int64Index([ 5, 11, 14, 17, 32, 41, 43, 50, 51, 54, 68, 69, 73, 74, 84, 109, 122, 130, 143, 144, 145, 174, 184], dtype='int64')

2) 在使用read_csv函数读取酒类消费数据表时（除文件地址外不添加额外的参数），pandas将continent字段中的“NA”（代表北美洲，North American）自动识别为NaN。因此，需要将continent字段中的NaN全部替换为字符串NA。如果学有余力，可以自行在网络上调研如何在read_csv函数中添加参数使NA不被识别为NaN；

？ >>> # 直接全部替换为’NA’
？ >>> df = df.fillna('NA')
？ >>> # 或者在读取数据时加入参数keep_default_na=False
？ >>> df = pd.read_csv('D:\drinks.csv', keep_default_na=False)

3) 分别输出各个大洲的平均啤酒、烈酒和红酒的消费量；

？ >>> df.groupby('continent')[['beer_servings', 'spirit_servings', 'wine_servings']].mean()
？ beer_servings  spirit_servings  wine_servings
？ continent
？ AF             61.471698        16.339623      16.264151
？ AS             37.045455        60.840909       9.068182
？ EU            193.777778       132.555556     142.222222
？ NA            145.434783       165.739130      24.521739
？ OC             89.687500        58.437500      35.625000
？ SA            175.083333       114.750000      62.416667

4) 分别输出啤酒、烈酒和红酒消费量最高的国家。

？ >>> df.iloc[df.beer_servings.argmax(), 0]
？ 'Namibia'
？ >>> df.iloc[df.spirit_servings.argmax(), 0]
？ 'Grenada'
？ >>> df.iloc[df.wine_servings.argmax(), 0]
？ 'France'

3. 狗狗币的历史价格

1) 用pandas将历史价格表中的数据读取为DataFrame，并查看各个列的数据类型。在读取数据时，pandas是否将表中的日期字段自动读取为日期类型？若否，则将其转换为日期类型；

？ >>> import pandas as pd
？ >>> df = pd.read_csv('D:\DOGE-USD.csv')
？ >>> df.dtypes                                                                                                           Date       object
？ Open      float64
？ High      float64
？ Low       float64
？ Close     float64
？ Volume    float64                                                                                                       dtype: object
？ >>> # 可见Date字段不是日期类型
？ >>> df.Date = pd.to_datetime(df.Date)
？ >>> df.dtypes
？ Date      datetime64[ns]
？ Open             float64
？ High             float64
？ Low              float64
？ Close            float64
？ Volume           float64
？ dtype: object

2) 该DataFrame中是否存在缺失值？若是，则输出数据缺失的日期，并用前一交易日的数据填充缺失值；

？ >>> df.info()
？ <class 'pandas.core.frame.DataFrame'>
？ RangeIndex: 2358 entries, 0 to 2357
？ Data columns (total 6 columns):
？ Date      2358 non-null datetime64[ns]
？ Open      2354 non-null float64
？ High      2354 non-null float64
？ Low       2354 non-null float64
？ Close     2354 non-null float64
？ Volume    2354 non-null float64
？ dtypes: datetime64[ns](1), float64(5)
？ memory usage: 110.7 KB
？ >>> # 可见除了Date字段外，其余字段都存在缺失值
？ >>> df['Date'].loc[df['Open'].isnull()]
？ 2039   2020-04-17
？ 2214   2020-10-09
？ 2217   2020-10-12
？ 2218   2020-10-13
？ Name: Date, dtype: datetime64[ns]
？ >>> df['Date'].loc[df['Close'].isnull()]
？ 2039   2020-04-17
？ 2214   2020-10-09
？ 2217   2020-10-12
？ 2218   2020-10-13
？ Name: Date, dtype: datetime64[ns]
？ >>> df['Date'].loc[df['High'].isnull()]
？ 2039   2020-04-17
？ 2214   2020-10-09
？ 2217   2020-10-12
？ 2218   2020-10-13
？ Name: Date, dtype: datetime64[ns]
？ >>> df['Date'].loc[df['Low'].isnull()]
？ 2039   2020-04-17
？ 2214   2020-10-09
？ 2217   2020-10-12
？ 2218   2020-10-13
？ Name: Date, dtype: datetime64[ns]
？ >>> df['Date'].loc[df['Volume'].isnull()]
？ 2039   2020-04-17
？ 2214   2020-10-09
？ 2217   2020-10-12
？ 2218   2020-10-13
？ Name: Date, dtype: datetime64[ns]
？ >>> # 可见存在缺失值的日期有四个
？ >>> df = df.fillna(method='pad')

3) 分别输出狗狗币价格的最高值与最低值，并分别输出达到最高值与最低值的日期；

>>> df.High.max()
？ 0.084945
？ >>> df.Low.min()
？ 8.499999999999999e-05
？ >>> df['Date'].loc[df['High'].argmax()]
？ Timestamp('2021-02-08 00:00:00')
？ >>> df['Date'].loc[df['Low'].argmin()]
？ Timestamp('2015-05-07 00:00:00')

4) 画出狗狗币每天最高价格的折线图（横轴为日期）；

？ >>> import matplotlib.pyplot as plt
？ >>> plt.plot(df.Date, df.High)
？ [<matplotlib.lines.Line2D object at 0x0000023AC49A0C48>]
？ >>> plt.show()  # 如图8-1所示

5) 画出狗狗币成交量的折线图（横轴为日期）。由于成交量字段中的数据数量级变化较大，直接画图无法直观地观察出其变化趋势，尝试画出更直观的成交量折线图（提示：取对数）。

？ >>> # 由于成交量的变化程度以及数量级较大
？ >>> # 因此考虑使用成交量的对数为纵轴
？ >>> plt.plot(df.Date, np.log10(df.Volume))
？ [<matplotlib.lines.Line2D object at 0x0000023AC49A0C48>]
？ >>> plt.show()  # 如图8-2示

转载请注明来源，欢迎对文章中的引用来源进行考证，欢迎指出任何有错误或不够清晰的表达。可以在下面评论区评论，也可以邮件至 1401362462@qq.com

exam2

网络爬虫

（一）显示影片基本信息