爬取链家租房信息

使用Python的requests和BeautifulSoup库爬取链家租房数据。以下是一个示例代码框架：

import requests
from bs4 import BeautifulSoup
import pandas as pd

def get_lianjia_rent_data(city='sh', max_pages=10):
    base_url = f"https://{city}.lianjia.com/zufang/"
    headers = {'User-Agent': 'Mozilla/5.0'}
    data = []
    
    for page in range(1, max_pages+1):
        url = f"{base_url}pg{page}"
        response = requests.get(url, headers=headers)
        soup = BeautifulSoup(response.text, 'html.parser')
        
        for item in soup.select('.content__list--item'):
            title = item.select_one('.content__list--item--title a').text.strip()
            price = item.select_one('.content__list--item-price').text.strip()
            area = item.select_one('.content__list--item--des a').text.strip()
            data.append([title, price, area])
    
    return pd.DataFrame(data, columns=['title', 'price', 'area'])

df = get_lianjia_rent_data()
df.to_csv('lianjia_rent.csv', index=False)

数据预处理

将爬取的数据进行清洗和转换：

# 价格提取数字
df['price'] = df['price'].str.extract('(\d+)').astype(float)

# 面积提取数字
df['area'] = df['area'].str.extract('(\d+)').astype(float)

# 去重和缺失值处理
df = df.drop_duplicates().dropna()

Hadoop环境搭建

搭建Hadoop集群需要以下组件：

• Hadoop 3.x
• HDFS分布式文件系统
• YARN资源管理器
• Java 8+

配置核心文件：

• core-site.xml
• hdfs-site.xml
• mapred-site.xml
• yarn-site.xml

数据导入HDFS

将清洗后的数据上传至HDFS：

hadoop fs -mkdir /input/lianjia
hadoop fs -put lianjia_rent.csv /input/lianjia

MapReduce分析

编写MapReduce程序分析租房价格分布：

public class RentAnalysis extends Configured implements Tool {
    
    public static class TokenizerMapper extends Mapper<Object, Text, Text, DoubleWritable>{
        private Text areaRange = new Text();
        private DoubleWritable price = new DoubleWritable();
        
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String[] fields = value.toString().split(",");
            if (fields.length >= 3) {
                double area = Double.parseDouble(fields[2]);
                double priceVal = Double.parseDouble(fields[1]);
                String range = getAreaRange(area);
                areaRange.set(range);
                price.set(priceVal);
                context.write(areaRange, price);
            }
        }
        
        private String getAreaRange(double area) {
            if (area < 50) return "0-50";
            else if (area < 100) return "50-100";
            else return "100+";
        }
    }
    
    public static class AvgReducer extends Reducer<Text, DoubleWritable, Text, DoubleWritable> {
        private DoubleWritable result = new DoubleWritable();
        
        public void reduce(Text key, Iterable<DoubleWritable> values, Context context) throws IOException, InterruptedException {
            double sum = 0;
            int count = 0;
            for (DoubleWritable val : values) {
                sum += val.get();
                count++;
            }
            result.set(sum / count);
            context.write(key, result);
        }
    }
    
    public int run(String[] args) throws Exception {
        Configuration conf = getConf();
        Job job = Job.getInstance(conf, "rent analysis");
        job.setJarByClass(RentAnalysis.class);
        job.setMapperClass(TokenizerMapper.class);
        job.setReducerClass(AvgReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(DoubleWritable.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        return job.waitForCompletion(true) ? 0 : 1;
    }
    
    public static void main(String[] args) throws Exception {
        int exitCode = ToolRunner.run(new RentAnalysis(), args);
        System.exit(exitCode);
    }
}

运行MapReduce作业

编译打包后提交作业：

hadoop jar RentAnalysis.jar /input/lianjia /output/rent_analysis

结果可视化

从HDFS获取结果并可视化：

import matplotlib.pyplot as plt

# 从HDFS获取结果
hadoop fs -get /output/rent_analysis/part-r-00000 results.txt

# 读取结果
with open('results.txt') as f:
    data = [line.strip().split('\t') for line in f]

ranges = [x[0] for x in data]
prices = [float(x[1]) for x in data]

# 绘制柱状图
plt.bar(ranges, prices)
plt.xlabel('Area Range (sqm)')
plt.ylabel('Average Price (yuan)')
plt.title('Average Rent Price by Area Range')
plt.show()

Hive数据分析

创建Hive表进行更复杂查询：

CREATE EXTERNAL TABLE lianjia_rent (
    title STRING,
    price DOUBLE,
    area DOUBLE
)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ','
STORED AS TEXTFILE
LOCATION '/input/lianjia';

-- 查询各区域平均价格
SELECT area, AVG(price) as avg_price
FROM lianjia_rent
GROUP BY area
ORDER BY avg_price DESC;

性能优化建议

使用Spark替代MapReduce可获得更好性能：

from pyspark.sql import SparkSession
from pyspark.sql.functions import avg, col

spark = SparkSession.builder.appName("LianjiaAnalysis").getOrCreate()
df = spark.read.csv("hdfs:///input/lianjia/lianjia_rent.csv", header=True, inferSchema=True)

result = df.groupBy("area").agg(avg("price").alias("avg_price")).orderBy("avg_price", ascending=False)
result.show()

result.write.csv("hdfs:///output/spark_rent_analysis")