### 使用Python编写的网络爬虫抓取文本数据并存储至Hadoop或Spark #### 抓取文本数据 为了创建一个能够有效抓取所需信息的Python爬虫，通常会利用`requests`库来发送HTTP请求以及`BeautifulSoup`库解析HTML页面。下面是一个简单的例子展示如何构建基本的爬虫： ```python import requests from bs4 import BeautifulSoup def fetch_webpage(url): response = requests.get(url) soup = BeautifulSoup(response.text, 'html.parser') paragraphs = [] for p in soup.find_all('p'): paragraphs.append(p.get_text()) return paragraphs ``` 这段代码定义了一个函数`fetch_webpage()`用于获取指定URL下的所有段落文字[^2]。 #### 数据预处理与清洗 在将收集到的信息存入分布式文件系统之前，可能还需要对这些原始数据做一些清理工作，比如去除空白符、标准化编码等。这一步骤有助于提高后续分析的质量和效率。 #### 存储至Hadoop Distributed File System (HDFS) 对于希望把大量非结构化数据保存下来供以后离线批处理的情况来说，可以选择使用Apache Hadoop作为底层存储解决方案之一。这里给出一段示范性的命令行指令说明怎样通过PyWebHdfs客户端上传本地文件到远程集群中的特定路径下： ```bash pip install pywebhdfs ``` ```python from pywebhdfs.webhdfs import PyWebHdfsClient hdfs = PyWebHdfsClient(host='namenode_host', port='50070', user_name='your_username') with open('/path/to/local/file.txt') as f: hdfs.create_file('/user/your_username/data/outputfile.txt',f.read()) ``` 上述代码片段展示了如何借助于`pywebhdfs`这个第三方库向HDFS写入新文件[^1]。 #### 利用Spark Streaming实时处理流式输入 如果应用场景涉及到持续不断地接收来自互联网的新鲜资讯更新，则可以考虑采用基于内存计算模型设计而成的大规模数据分析引擎——Apache Spark来进行近似即时的数据流转播送服务。以下是有关设置Kafka-Spark集成环境的部分要点概述： 安装必要的依赖项： ```bash pip install pyspark kafka-python ``` 编写消费者逻辑读取消息队列里的项目，并将其传递给RDD对象以便进一步转换操作： ```python from pyspark.sql import SparkSession from pyspark.streaming.kafka import KafkaUtils from pyspark.streaming import StreamingContext spark_session = SparkSession.builder.appName("StreamingApp").getOrCreate() streaming_context = StreamingContext(spark_session.sparkContext, batch_duration=1) kvs = KafkaUtils.createDirectStream(streaming_context, ['topic-name'], {'metadata.broker.list': 'localhost:9092'}) lines = kvs.map(lambda x: x[1]) words = lines.flatMap(lambda line: line.split(" ")) word_counts = words.countByValue() word_counts.pprint() # 打印结果 streaming_context.start() streaming_context.awaitTermination() ``` 此部分介绍了如何连接到Kafka主题订阅消息源并通过调用内置方法统计词频分布情况。