startUrl=r'http://ccs.snnu.edu.cn/xygk/lsyg1.htm' with urlopen(startUrl) as fp: content=fp.read().decode() #提取并遍历每个事件链接 pattern=re.compile(u'<p.*?<span style="background.*?>(.*?)</span>.*?' +'<span style="background.*?>:(.*?)</span>(.*?)</span></p>',re.I) result=re.findall(pattern,content) file_test=open('test_example','w',encoding='utf-8')

### 实现方案 以下是基于提供的引用内容和专业知识设计的一个完整的解决方案，涵盖了从Kafka主题订阅到数据处理再到MySQL存储的过程。 #### 1. 使用 `Assign` 策略订阅指定分区的主题 为了使用 `Assign` 策略订阅特定的分区，可以通过 `ConsumerStrategies.Assign()` 方法显式定义要消费的分区及其对应的起始偏移量。这种方式适用于需要精确控制哪些分区被消费的情况[^3]。 ```scala import org.apache.kafka.common.TopicPartition import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent import org.apache.spark.streaming.kafka010._ import org.apache.spark.streaming.{Seconds, StreamingContext} import org.apache.spark.SparkConf val kafkaParams = Map[String, Object]( "bootstrap.servers" -> "localhost:9092", "key.deserializer" -> classOf[org.apache.kafka.common.serialization.StringDeserializer], "value.deserializer" -> classOf[org.apache.kafka.common.serialization.StringDeserializer], "group.id" -> "use_a_separate_group_id_for_each_stream", "auto.offset.reset" -> "earliest", "enable.auto.commit" -> (false: java.lang.Boolean) ) // 定义TopicPartitions val topicPartitions = Seq( new TopicPartition("test-topic", 0), new TopicPartition("test-topic", 1) ) // 起始偏移量 val fromOffsets = topicPartitions.map(tp => tp -> 0L).toMap // 创建DStream val stream = KafkaUtils.createDirectStream[ String, String ](ssc, PreferConsistent, Assign[String, String](topicPartitions, kafkaParams, fromOffsets)) ``` --- #### 2. 数据解析与清洗 对于流式数据的解析和清洗操作，可以利用 Spark 的 DStream 提供的操作接口完成时间戳转换、URL 提取等任务。 ```scala stream.foreachRDD { rdd => val data = rdd.map(record => { val value = record.value() try { // 假设每条消息是一个JSON对象 import org.json4s._ import org.json4s.jackson.JsonMethods._ implicit val formats = DefaultFormats val json = parse(value) // 时间戳转换为字符串 val timestamp = (json \ "timestamp").extractOpt[Long].map(_.toString).getOrElse("") // URL提取 val url = (json \ "url").extractOpt[String].getOrElse("") // 结果组装 s"$timestamp,$url" } catch { case e: Exception => "" } }) // 进行非空过滤 val cleanedData = data.filter(!_.isEmpty()) // 将清理后的数据保存至MySQL saveToMySQL(cleanedData.collect().toList) } def saveToMySQL(dataList: List[String]): Unit = { dataList.foreach(row => { // 解析row中的字段并执行SQL插入语句 val Array(timestamp, url) = row.split(",") Class.forName("com.mysql.cj.jdbc.Driver") val connection = java.sql.DriverManager.getConnection("jdbc:mysql://localhost:3306/testdb?user=root&password=pass") val statement = connection.prepareStatement("INSERT INTO logs (timestamp, url) VALUES (?, ?)") statement.setString(1, timestamp) statement.setString(2, url) statement.executeUpdate() statement.close() connection.close() }) } ``` --- #### 3. MySQL 存储配置 在上述代码中，假设已经有一个名为 `logs` 的表用于存储处理后的数据。该表结构如下： ```sql CREATE TABLE logs ( id INT AUTO_INCREMENT PRIMARY KEY, timestamp VARCHAR(50), url TEXT ); ``` 需要注意的是，每次连接数据库可能会带来性能开销。因此，在实际生产环境中建议批量提交数据或将 JDBC 操作封装成更高效的工具类[^2]。 --- ### 总结 以上代码实现了以下几个功能： - **Kafka 主题订阅**：通过 `Assign` 策略指定了具体的分区和起始偏移量。 - **数据解析与清洗**：完成了 JSON 数据的时间戳转换、URL 提取以及结果组装。 - **MySQL 存储**：将清洗后的数据逐条或批量写入 MySQL 表中。 --- ###

在Python中，你可以创建一个名为`crawler`的函数，该函数接受两个参数：`startURL`和`dataSavPath`，用于网络爬取和数据存储。这个函数通常用于编写网络爬虫程序，如使用`requests`库获取网页内容，然后使用如`BeautifulSoup`或`lxml`解析HTML，最后将数据保存到指定路径的文件中。以下是一个基本的框架示例： ```python import requests from bs4 import BeautifulSoup import os def crawler(start_URL, data_save_path): # 检查数据保存路径是否存在，不存在则创建 if not os.path.exists(os.path.dirname(data_save_path)): os.makedirs(os.path.dirname(data_save_path)) # 发送HTTP请求 try: response = requests.get(start_URL, timeout=10) # 设置超时时间防止阻塞 response.raise_for_status() # 如果状态码不是200，抛出异常 except (requests.exceptions.RequestException, ValueError) as e: print(f"Error occurred while fetching URL: {e}") return # 解析HTML soup = BeautifulSoup(response.text, 'html.parser') # 这里根据实际需求提取所需信息，例如标题、链接等 title = soup.find('title').get_text() links = [a['href'] for a in soup.select('a[href]')] # 示例：提取所有链接 # 将数据保存到文件 with open(data_save_path, 'w', encoding='utf-8') as f: for link in links: f.write(f"{link}\n") f.write(f"\nTitle: {title}") print(f"Crawled and saved data to: {data_save_path}") # 使用示例 start_url = "http://example.com" # 替换为你想爬取的URL save_path = "data/crawled_links.txt" # 数据保存路径 crawler(start_url, save_path)