项目六 动态网页数据预处理

pd.read_csv(filepath_or_buffer, sep=',', header='infer', names=None, index_col=None,
            prefix=None, dtype=None, encoding=None, converters=None,
            skipinitialspace=False, na_values=None, na_filter=True,
            true_values=None, false_values=None)

pd.read_excel(io, sheet_name=0, header=0, names=None, index_col=None, usecols=None,
              squeeze=False, dtype=None, true_values=None, false_values=None,
              skiprows=None, nrows=None)

pandas.read_sql(sql, con, index_col=None, coerce_float=True, parse_dates=None,
                columns=None, chunksize=None)

pd.read_json(path_or_buf=None, orient=None, typ='frame', dtype=True,
             keep_default_dates=True, numpy=False, date_unit=None,
             encoding=None, lines=False)

DataFrame.to_csv(path_or_buf=None, sep=',', na_rep='', float_format=None,
                 columns=None, header=True, index=True, index_label=None,
                 mode='w', encoding=None)

DataFrame.to_excel(excel_writer, sheet_name='Sheet1', na_rep='',
                   float_format=None, columns=None, header=True, index=True,
                   startrow=0, startcol=0)

DataFrame.to_json(path_or_buf=None, orient=None, date_format=None,
                  double_precision=10, force_ascii=True, date_unit='ms',
                  default_handler=None, lines=False, compression=None, index=True)

DataFrame.to_sql(name, con, schema=None, if_exists='fail', index=True,
                 index_label=None, chunksize=None)

df.isna()

df.isna().sum(axis=None)

df.fillna(value=None, method=None, axis=None,
          inplace=False, limit=None, downcast=None)

df.dropna(axis=None)

# 配置环境变量 ~/.bashrc
[root@master local]# vim ~/.bashrc
export PIG_HOME=/usr/local/pig
export PATH=$PATH:$PIG_HOME/bin
export PIG_CLASSPATH=$HADOOP_HOME/etc/hadoop
[root@master local]# source ~/.bashrc
[root@master local]# pig -version

Relation_name = LOAD 'Input file path' USING function as schema;

STORE Relation_name INTO 'required_directory_path' [USING function];

# 对单个关系分组
Group_data = GROUP Relation_name BY Group_key;
# 对多个关系分组
Group_data = GROUP Relation_name1 BY Group_key, Relation_name2 BY Group_key;

# 自连接
Relation_name = JOIN Relation1_name BY key, Relation2_name BY key;

# 内部连接
result = JOIN relation1 BY columnname, relation2 BY columnname;

# 左外连接
outer_left = JOIN relation1 BY columnname LEFT, relation2 BY columnname;

# 右外连接
outer_right = JOIN relation1 BY columnname RIGHT, relation2 BY columnname;

# 全外连接
outer_full = JOIN relation1 BY columnname FULL OUTER, relation2 BY columnname;

input {   # 输入组件配置
}
filter {  # 过滤组件配置
}
output {  # 输出组件配置
}

logstash -f ./logstash.conf

# pandas_clean.py
import pandas as pd

# 1. 读取采集到的新闻数据
df = pd.read_csv("news_raw.csv")

# 2. 检查基本信息与缺失情况
print("维度：", df.shape)
print("列名：", list(df.columns))
print("缺失统计：\n", df.isna().sum())

# 3. 去重：按"标题"去重，保留第一条
df = df.drop_duplicates(subset=["标题"])

# 4. 填充：标题为空用"未知"，阅读量为空用 0
df["标题"] = df["标题"].fillna("未知")
df["阅读量"] = df["阅读量"].fillna(0)

# 5. 类型转换：阅读量转为整数
df["阅读量"] = df["阅读量"].astype(int)

# 6. 保存清洗结果
df.to_csv("news_clean.csv", index=False, encoding="utf-8-sig")
print("清洗后剩余：", len(df), "条")

维度： (120, 5)
列名： ['标题', '发布时间', '来源', '链接', '阅读量']
缺失统计：
 标题      2
发布时间    0
来源      0
链接      0
阅读量     1
清洗后剩余： 112 条

# pandas_groupby.py
import pandas as pd

df = pd.read_csv("news_clean.csv")

# 按"来源"分组，聚合新闻条数与平均阅读量
result = df.groupby("来源").agg(
    条数=("标题", "count"),
    平均阅读量=("阅读量", "mean")
).reset_index()

# 按条数降序排列
result = result.sort_values("条数", ascending=False)
print(result)

# 输出各来源占比
total = result["条数"].sum()
for _, row in result.iterrows():
    print(f"{row['来源']}：{row['条数']} 条，占比 {row['条数']/total:.1%}")

     来源  条数   平均阅读量
0  官方发布    45  1325.6
1  校园新闻    38   876.4
2  行业资讯    29  1102.3
官方发布：45 条，占比 40.2%
校园新闻：38 条，占比 33.9%
行业资讯：29 条，占比 25.9%

# logstash.conf —— 解析 Nginx 访问日志
input {
  kafka {
    bootstrap_servers => "localhost:9092"
    topics => ["nginx-log"]
    codec => "json"
  }
}

filter {
  grok {
    # 匹配：客户端IP 时间 "请求" 状态码 字节 来源
    match => { "message" => "%{IP:client_ip} \[%{HTTPDATE:time}\] \"%{WORD:method} %{URIPATHPARAM:request}\" %{INT:status} %{INT:bytes} \"%{DATA:referer}\"" }
  }
  date {
    match => ["time", "dd/MMM/yyyy:HH:mm:ss Z"]
    target => "@timestamp"
  }
  geoip {
    source => "client_ip"
  }
}

output {
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "inspur-logdata-2022"
  }
  stdout { codec => rubydebug }
}

# 启动
[root@master logstash]# bin/logstash -f logstash.conf

# 输入日志样例
192.168.1.10 [10/Feb/2022:10:15:30 +0800] "GET /index.html HTTP/1.1" 200 5120 "-"

# 解析后的结构化字段
client_ip : 192.168.1.10
method    : GET
request   : /index.html
status    : 200
bytes     : 5120
@timestamp: 2022-02-10T02:15:30.000Z