Hadoop

本文集文档发布于灏天文库

3.2 MapReduce API 与编程实践

文档摘要

3.2 MapReduce API 与编程实践 第三章：Hadoop MapReduce 分布式计算框架 3.2 MapReduce API 与编程实践 3.2.1 MapReduce API 核心概念 MapReduce 框架的核心思想来源于函数式编程中的 Map 和 Reduce 操作。它将复杂的数据处理任务分解成两个主要的阶段：Map 阶段和 Reduce 阶段。这两个阶段通过键值对 (Key-Value Pair) 数据结构进行数据传递和处理。 1. 输入与输出类型：键值对 (Key-Value Pair) MapReduce 框架处理的数据都以键值对的形式存在。无论是输入数据、中间数据还是最终输出数据，都是键值对的形式。

3.2 MapReduce API 与编程实践

第三章：Hadoop MapReduce 分布式计算框架

3.2 MapReduce API 与编程实践

3.2.1 MapReduce API 核心概念

MapReduce 框架的核心思想来源于函数式编程中的 Map 和 Reduce 操作。它将复杂的数据处理任务分解成两个主要的阶段：Map 阶段和 Reduce 阶段。这两个阶段通过键值对 (Key-Value Pair) 数据结构进行数据传递和处理。

1. 输入与输出类型：键值对 (Key-Value Pair)

MapReduce 框架处理的数据都以键值对的形式存在。无论是输入数据、中间数据还是最终输出数据，都是键值对的形式。键 (Key) 和值 (Value) 可以是任意类型，但通常需要是 Hadoop Writable 接口的实现类，以便于序列化和反序列化，从而在分布式环境中进行数据传输和存储。

2. Map 阶段

Map 阶段是整个 MapReduce 流程的第一个阶段。它接收输入数据，并将其转换为一系列的中间键值对。Map 阶段由 Mapper 组件负责执行。

  • Mapper 的作用：

    • 读取输入数据：Mapper 从 InputFormat 组件获取输入数据分片 (InputSplit)，并逐条读取记录。

    • 数据转换：Mapper 对每条输入记录进行处理，将其转换为一个或多个中间键值对。这个转换过程通常涉及到数据的解析、过滤、转换等操作，根据具体的业务逻辑而定。

    • 输出中间结果：Mapper 将生成的中间键值对输出，这些中间结果将作为 Reduce 阶段的输入。

  • Mapper 接口：

    public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {
    // ...
    protected void map(KEYIN key, VALUEIN value, Context context) throws IOException, InterruptedException {
        // 核心的 Map 处理逻辑
        // 使用 context.write(KEYOUT key, VALUEOUT value) 输出中间键值对
    }
    // ...
}
​

    • KEYIN, VALUEIN: Mapper 输入键值对的类型。由 InputFormat 决定。

    • KEYOUT, VALUEOUT: Mapper 输出键值对的类型，作为 Reduce 阶段的输入类型。

    • map(KEYIN key, VALUEIN value, Context context) 方法：是 Mapper 的核心方法，框架会为每个输入键值对调用一次该方法。

    • Context context: 上下文对象，用于输出中间结果、获取配置信息等。context.write(KEYOUT key, VALUEOUT value) 方法用于输出中间键值对。

3. Shuffle 阶段

Shuffle 阶段是 Map 阶段和 Reduce 阶段之间的桥梁，也是 MapReduce 框架中最复杂、最关键的阶段。Shuffle 阶段负责对 Map 阶段输出的中间结果进行分区 (Partition)、排序 (Sort)、合并 (Combine) 等操作，并将处理后的数据分发到不同的 Reducer 节点。

  • Partition (分区)： 根据中间结果的键 (Key)，将数据划分到不同的分区。每个分区的数据将由一个 Reducer 实例处理。默认的分区器是 HashPartitioner，它根据键的哈希值对 Reducer 的数量取模来决定数据属于哪个分区。用户也可以自定义分区器，以实现更灵活的数据分配策略。

  • Sort (排序)： 对每个分区内的数据按照键 (Key) 进行排序。排序是 Shuffle 阶段的默认操作，它可以保证相同键的数据会被分发到同一个 Reducer，并且在 Reducer 端按照键的顺序进行处理。

  • Combine (合并)： 可选的优化步骤。Combiner 的作用是在 Map 阶段输出的中间结果进行本地聚合，减少网络传输的数据量，提高效率。Combiner 的逻辑通常与 Reducer 的逻辑相似，但 Combiner 是在 Map 节点本地执行的，而 Reducer 是在 Reduce 节点执行的。

4. Reduce 阶段

Reduce 阶段是 MapReduce 流程的第二个阶段。它接收 Shuffle 阶段处理后的数据，并对相同键的数据进行聚合、汇总等操作，最终生成最终的输出结果。Reduce 阶段由 Reducer 组件负责执行。

  • Reducer 的作用：

    • 接收中间结果：Reducer 从 Shuffle 阶段接收属于自己分区的数据，这些数据已经按照键排序。

    • 数据聚合：Reducer 对接收到的数据进行聚合、汇总等操作，根据业务逻辑，将相同键的值进行处理，生成最终的输出结果。

    • 输出最终结果：Reducer 将处理后的最终结果输出到 OutputFormat 组件。

  • Reducer 接口：

    public class Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {
    // ...
    protected void reduce(KEYIN key, Iterable<VALUEIN> values, Context context) throws IOException, InterruptedException {
        // 核心的 Reduce 处理逻辑
        // 使用 context.write(KEYOUT key, VALUEOUT value) 输出最终结果
    }
    // ...
}
​

    • KEYIN, VALUEIN: Reducer 输入键值对的类型，与 Mapper 的输出类型一致。

    • KEYOUT, VALUEOUT: Reducer 输出键值对的类型，作为最终输出结果的类型。

    • reduce(KEYIN key, Iterable<VALUEIN> values, Context context) 方法：是 Reducer 的核心方法，框架会为每个键调用一次该方法。values 参数是一个迭代器，包含了所有具有相同键的值。

    • Context context: 上下文对象，用于输出最终结果、获取配置信息等。context.write(KEYOUT key, VALUEOUT value) 方法用于输出最终结果键值对。

5. InputFormat 与 OutputFormat

InputFormat 和 OutputFormat 组件负责处理 MapReduce 作业的输入和输出数据。

  • InputFormat：

    • 作用：

      • 数据读取：InputFormat 负责从输入源（如 HDFS 文件）读取数据，并将数据划分为 InputSplit，供 Mapper 处理。

      • 数据格式化：InputFormat 将读取的数据解析成键值对形式，供 Mapper 使用。

    • 常见的 InputFormat 实现类：

      • TextInputFormat: 默认的文本文件输入格式，按行读取文本文件，键为每行在文件中的字节偏移量 (LongWritable)，值为每行文本内容 (Text)。

      • KeyValueTextInputFormat: 适用于每行都是键值对的文本文件，键和值之间用分隔符分隔。

      • NLineInputFormat: 按 N 行分割输入文件。

      • FileInputFormat: 所有基于文件的 InputFormat 的基类。

      • SequenceFileInputFormat: 读取 SequenceFile 格式的文件。

      • AvroFileInputFormat, ParquetInputFormat, ORCInputFormat: 用于读取特定格式的文件。

  • OutputFormat：

    • 作用：

      • 数据输出：OutputFormat 负责将 Reducer 的输出结果写入到指定的输出目标（如 HDFS 文件）。

      • 数据格式化：OutputFormat 将 Reducer 输出的键值对格式化为指定的输出格式。

    • 常见的 OutputFormat 实现类：

      • TextOutputFormat: 默认的文本文件输出格式，将键值对以文本形式写入文件，键和值之间用制表符分隔。

      • FileOutputFormat: 所有基于文件的 OutputFormat 的基类。

      • SequenceFileOutputFormat: 将输出结果写入 SequenceFile 格式的文件。

      • AvroFileOutputFormat, ParquetOutputFormat, ORCOutputFormat: 用于输出特定格式的文件。

      • MultipleOutputs: 允许作业输出到多个不同的输出目标或不同的格式。

MapReduce 数据处理流程图 (Mermaid Graph)：

3.2.2 MapReduce 编程实践：WordCount 示例

WordCount 是 MapReduce 编程中最经典的入门示例，它可以统计文本文件中每个单词出现的次数。通过 WordCount 示例，我们可以深入理解 MapReduce API 的使用方法和编程流程。

1. 代码实现 (Java):

import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
public class WordCount {
    // Mapper 类：将每行文本拆分成单词，并输出 (word, 1) 键值对
    public static class TokenizerMapper
            extends Mapper<Object, Text, Text, IntWritable> {
        private final static IntWritable one = new IntWritable(1); // 定义常量 1
        private Text word = new Text(); // 可重用的 Text 对象，避免频繁创建
        @Override
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            StringTokenizer itr = new StringTokenizer(value.toString()); // 使用 StringTokenizer 分词
            while (itr.hasMoreTokens()) {
                word.set(itr.nextToken()); // 设置单词
                context.write(word, one); // 输出 (word, 1) 键值对
            }
        }
    }
    // Reducer 类：统计相同单词的出现次数
    public static class IntSumReducer
            extends Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable result = new IntWritable(); // 可重用的 IntWritable 对象
        @Override
        public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable val : values) { // 遍历相同单词的所有计数
                sum += val.get(); // 累加计数
            }
            result.set(sum); // 设置总计数
            context.write(key, result); // 输出 (word, sum) 键值对
        }
    }
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration(); // 创建 Hadoop 配置对象
        Job job = Job.getInstance(conf, "word count"); // 创建 Job 对象，并设置作业名称
        job.setJarByClass(WordCount.class); // 设置 Job 的主类
        job.setMapperClass(TokenizerMapper.class); // 设置 Mapper 类
        job.setCombinerClass(IntSumReducer.class); // 设置 Combiner 类 (可选，这里使用 Reducer 作为 Combiner)
        job.setReducerClass(IntSumReducer.class); // 设置 Reducer 类
        job.setOutputKeyClass(Text.class); // 设置输出键类型
        job.setOutputValueClass(IntWritable.class); // 设置输出值类型
        FileInputFormat.addInputPath(job, new Path(args[0])); // 设置输入路径
        FileOutputFormat.setOutputPath(job, new Path(args[1])); // 设置输出路径
        System.exit(job.waitForCompletion(true) ? 0 : 1); // 提交 Job 并等待完成，根据结果码退出
    }
}
​

2. 代码详解：

  • Mapper 类 ( TokenizerMapper )：

    • 继承 Mapper<Object, Text, Text, IntWritable>：

      • Object: 输入键类型，TextInputFormat 默认的键类型是 LongWritable (字节偏移量)，这里使用 Object 可以兼容。

      • Text: 输入值类型，TextInputFormat 默认的值类型是 Text (每行文本)。

      • Text: 输出键类型，Mapper 输出的键是单词，类型为 Text。

      • IntWritable: 输出值类型，Mapper 输出的值是计数 1，类型为 IntWritable。

    • map(Object key, Text value, Context context) 方法：

      • value.toString(): 将输入的 Text 值转换为 String 类型。

      • StringTokenizer itr = new StringTokenizer(value.toString()): 使用 StringTokenizer 对每行文本进行分词，默认以空格、制表符、换行符等作为分隔符。

      • while (itr.hasMoreTokens()): 遍历每个单词。

      • word.set(itr.nextToken()): 获取下一个单词，并设置到可重用的 Text 对象 word 中。

      • context.write(word, one): 将单词和计数 1 作为键值对输出。context.write() 方法会将数据写入到 Shuffle 阶段。

  • Reducer 类 ( IntSumReducer )：

    • 继承 Reducer<Text, IntWritable, Text, IntWritable>：

      • Text: 输入键类型，与 Mapper 的输出键类型一致，为 Text (单词)。

      • IntWritable: 输入值类型，与 Mapper 的输出值类型一致，为 IntWritable (计数 1)。

      • Text: 输出键类型，Reducer 输出的键是单词，类型为 Text。

      • IntWritable: 输出值类型，Reducer 输出的值是总计数，类型为 IntWritable。

    • reduce(Text key, Iterable<IntWritable> values, Context context) 方法：

      • key: 输入键，即单词。框架会将具有相同键的所有键值对聚合到同一个 Reducer 的 reduce 方法中，并以键的形式传入。

      • Iterable<IntWritable> values: 输入值迭代器，包含了所有具有相同键 (单词) 的值，即 Mapper 输出的计数 1 的集合。

      • for (IntWritable val : values): 遍历所有计数。

      • sum += val.get(): 累加计数。

      • result.set(sum): 将总计数设置到可重用的 IntWritable 对象 result 中。

      • context.write(key, result): 将单词和总计数作为键值对输出。context.write() 方法会将数据写入到 OutputFormat 组件。

  • Main 函数 ( main )：

    • Configuration conf = new Configuration(): 创建 Hadoop 配置对象，用于加载 Hadoop 配置文件。

    • Job job = Job.getInstance(conf, "word count"): 创建 Job 对象，并设置作业名称为 "word count"。

    • job.setJarByClass(WordCount.class): 设置 Job 的主类，框架会根据主类找到包含 Mapper 和 Reducer 类的 Jar 包。

    • job.setMapperClass(TokenizerMapper.class): 设置 Mapper 类为 TokenizerMapper。

    • job.setCombinerClass(IntSumReducer.class): 设置 Combiner 类为 IntSumReducer。这里使用 Reducer 类作为 Combiner，因为 WordCount 的 Combiner 逻辑与 Reducer 逻辑相同，都是求和。Combiner 可以减少 Map 阶段输出到 Reduce 阶段的网络传输量。

    • job.setReducerClass(IntSumReducer.class): 设置 Reducer 类为 IntSumReducer。

    • job.setOutputKeyClass(Text.class): 设置输出键类型为 Text。

    • job.setOutputValueClass(IntWritable.class): 设置输出值类型为 IntWritable。

    • FileInputFormat.addInputPath(job, new Path(args[0])): 设置输入路径，从命令行参数获取。

    • FileOutputFormat.setOutputPath(job, new Path(args[1])): 设置输出路径，从命令行参数获取。

    • System.exit(job.waitForCompletion(true) ? 0 : 1): 提交 Job 到 Hadoop 集群运行，并等待作业完成。job.waitForCompletion(true) 方法会打印作业的运行进度和日志信息。根据作业的执行结果 (成功或失败) 返回相应的退出码 (0 或 1)。

3. 编译和运行 WordCount 作业：

  • 编译： 将 WordCount.java 文件编译成 .class 文件，并打包成 Jar 文件 (例如 wordcount.jar)。

    javac WordCount.java
jar cvf wordcount.jar WordCount*.class
​

  • 运行 (本地模式)： 在本地模式下运行 MapReduce 作业，方便调试和测试。

    hadoop jar wordcount.jar WordCount input.txt output_dir
​

    • hadoop jar wordcount.jar: 运行 Hadoop Jar 包的命令。

    • WordCount: 主类名。

    • input.txt: 输入文件路径 (本地文件系统)。

    • output_dir: 输出目录路径 (本地文件系统)，需要确保该目录不存在。

  • 运行 (集群模式)： 在 Hadoop 集群上运行 MapReduce 作业。需要将 wordcount.jar 上传到集群，并将输入文件上传到 HDFS。

    hadoop jar wordcount.jar WordCount hdfs://namenode:port/input_path hdfs://namenode:port/output_path
​

    • hdfs://namenode:port/input_path: HDFS 输入文件路径。

    • hdfs://namenode:port/output_path: HDFS 输出目录路径，需要确保该目录不存在。

4. WordCount 作业执行流程 (Mermaid Graph)：

流程说明：

  1. InputFormat (TextInputFormat)： TextInputFormat 读取输入文本文件，将文件分割成 InputSplit，并按行读取数据，输出键为字节偏移量 (Object)，值为每行文本 (Text)。

  2. Map 阶段 (TokenizerMapper)： TokenizerMapper 对每行文本进行分词，将每个单词转换为键 (Text)，值为 1 (IntWritable) 的键值对。例如，输入 "Hello World"，Mapper 会输出 ("Hello", 1) 和 ("World", 1)。

  3. Shuffle 阶段： Shuffle 阶段根据键 (单词) 将 Mapper 输出的键值对进行分区、排序。相同单词的键值对会被分发到同一个 Reducer 节点，并按照单词排序。

  4. Reduce 阶段 (IntSumReducer)： IntSumReducer 接收 Shuffle 阶段分发的数据，对相同单词的值 (计数 1) 进行累加求和，得到每个单词的总计数。例如，如果 Reducer 接收到 ("Hello", [1, 1, 1])，它会计算总和为 3，并输出 ("Hello", 3)。

  5. OutputFormat (TextOutputFormat)： TextOutputFormat 将 Reducer 输出的键值对以文本形式写入到输出文件，默认键和值之间用制表符分隔。

3.2.3 MapReduce 编程实践：Combiner 的使用

在 WordCount 示例中，我们使用了 Combiner。Combiner 的作用是在 Map 阶段输出的中间结果进行本地聚合，减少网络传输的数据量，提高效率。

1. Combiner 的必要性：

在 WordCount 示例中，如果输入文件非常大，Mapper 可能会输出大量的 (word, 1) 键值对。这些中间结果需要通过网络传输到 Reducer 节点，网络传输会成为性能瓶颈。Combiner 可以在 Map 节点本地先对相同单词的计数进行求和，减少网络传输的数据量。

2. Combiner 的使用：

在 WordCount 示例中，我们直接将 Reducer 类 (IntSumReducer) 设置为 Combiner 类：

job.setCombinerClass(IntSumReducer.class);
​

由于 WordCount 的 Combiner 逻辑与 Reducer 逻辑相同 (都是求和)，因此可以直接使用 Reducer 类作为 Combiner 类。

3. Combiner 的执行位置和次数：

  • Combiner 在每个 Map Task 的输出数据上执行。

  • Combiner 是可选的，不一定会执行。框架可以根据数据量、集群负载等情况决定是否执行 Combiner，以及执行多少次。

  • Combiner 的输入和输出类型必须与 Mapper 的输出类型和 Reducer 的输入类型一致。

4. Combiner 的局限性：

  • Combiner 只能用于满足特定条件的操作，例如求和、求最大值、求最小值等，这些操作满足结合律和交换律，可以在本地先进行部分聚合，再进行全局聚合。

  • 对于一些复杂的聚合操作，例如求平均值、去重计数等，Combiner 可能无法直接使用，需要进行特殊设计。

5. Combiner 的优势：

  • 减少网络传输量，提高 MapReduce 作业的执行效率。

  • 特别是在数据倾斜的情况下，Combiner 的作用更加明显。

3.2.4 MapReduce 编程实践：Partitioner 的使用

Partitioner 组件负责将 Map 阶段输出的中间结果划分到不同的分区，每个分区的数据将由一个 Reducer 实例处理。默认的分区器是 HashPartitioner，它根据键的哈希值对 Reducer 的数量取模来决定数据属于哪个分区。

1. 默认分区器 (HashPartitioner)：

HashPartitioner 的分区策略如下：

public class HashPartitioner<K, V> extends Partitioner<K, V> {
    public int getPartition(K key, V value, int numReduceTasks) {
        return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
    }
}
​

  • key.hashCode(): 获取键的哈希值。

  • & Integer.MAX_VALUE: 保证哈希值为正数。

  • % numReduceTasks: 对 Reducer 的数量取模，得到分区号 (0 到 numReduceTasks-1)。

HashPartitioner 的优点是分区均匀，可以保证数据在 Reducer 之间均匀分布，避免数据倾斜。但缺点是它只根据键的哈希值进行分区，无法根据业务逻辑进行灵活的分区控制。

2. 自定义 Partitioner：

用户可以自定义 Partitioner，以实现更灵活的数据分配策略。自定义 Partitioner 需要继承 Partitioner 类，并重写 getPartition() 方法。

示例：根据单词首字母进行分区

假设我们需要将单词按照首字母划分到不同的分区，首字母为 A-M 的单词分到分区 0，首字母为 N-Z 的单词分到分区 1。

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;
public class FirstLetterPartitioner extends Partitioner<Text, IntWritable> {
    @Override
    public int getPartition(Text key, IntWritable value, int numReduceTasks) {
        char firstLetter = key.toString().toUpperCase().charAt(0);
        if (firstLetter >= 'A' && firstLetter <= 'M') {
            return 0; // 分区 0
        } else {
            return 1 % numReduceTasks; // 分区 1 (如果 Reducer 数量大于 2，则取模)
        }
    }
}
​

3. 使用自定义 Partitioner：

在 Job 配置中设置自定义 Partitioner 类：

job.setPartitionerClass(FirstLetterPartitioner.class);
job.setNumReduceTasks(2); // 设置 Reducer 数量为 2
​

  • job.setPartitionerClass(FirstLetterPartitioner.class): 设置 Partitioner 类为 FirstLetterPartitioner。

  • job.setNumReduceTasks(2): 设置 Reducer 数量为 2，与分区数量一致。如果分区数量与 Reducer 数量不一致，可能会导致数据丢失或分区不均匀。

4. Partitioner 的作用：

  • 控制数据分发：Partitioner 可以控制 Map 阶段输出的中间结果如何分发到不同的 Reducer 节点。

  • 实现数据隔离：可以将不同类型的数据划分到不同的分区，由不同的 Reducer 处理，实现数据隔离和并行处理。

  • 优化负载均衡：自定义 Partitioner 可以根据数据分布情况进行优化，尽量保证数据在 Reducer 之间均匀分布，避免数据倾斜。

3.2.5 MapReduce 编程实践：InputFormat 和 OutputFormat 的选择

选择合适的 InputFormat 和 OutputFormat 对于 MapReduce 作业的性能和功能至关重要。

1. InputFormat 的选择：

  • 根据输入数据格式选择：

    • 文本文件：TextInputFormat, KeyValueTextInputFormat, NLineInputFormat

    • SequenceFile 文件：SequenceFileInputFormat

    • Avro 文件：AvroFileInputFormat

    • Parquet 文件：ParquetInputFormat

    • ORC 文件：ORCInputFormat

    • 数据库：DBInputFormat (需要配置数据库连接信息)

  • 根据数据分割方式选择：

    • 按行分割：TextInputFormat, KeyValueTextInputFormat

    • 按 N 行分割：NLineInputFormat

    • 自定义分割：可以自定义 InputFormat，实现更灵活的数据分割策略。

  • 考虑性能和效率：

    • 对于大文件，选择支持数据压缩的 InputFormat 可以减少磁盘 I/O 和网络传输。

    • 对于特定格式的文件，选择专用的 InputFormat 可以提高数据读取效率。

2. OutputFormat 的选择：

  • 根据输出数据格式选择：

    • 文本文件：TextOutputFormat

    • SequenceFile 文件：SequenceFileOutputFormat

    • Avro 文件：AvroFileOutputFormat

    • Parquet 文件：ParquetFileOutputFormat

    • ORC 文件：ORCOutputFormat

    • 数据库：DBOutputFormat (需要配置数据库连接信息)

  • 根据输出数据组织方式选择：

    • 单个输出文件：FileOutputFormat 的子类 (默认)

    • 多个输出文件：MultipleOutputs (可以根据键或值将数据输出到不同的文件)

  • 考虑性能和效率：

    • 选择支持数据压缩的 OutputFormat 可以减少磁盘空间占用。

    • 对于特定格式的文件，选择专用的 OutputFormat 可以提高数据写入效率。

3. 自定义 InputFormat 和 OutputFormat：

在一些特殊场景下，可能需要自定义 InputFormat 和 OutputFormat 来满足特定的需求。自定义 InputFormat 需要继承 InputFormat 类，并实现数据分割和记录读取逻辑。自定义 OutputFormat 需要继承 OutputFormat 类，并实现数据写入逻辑。

发布者: 作者: 转发
评论区 (0)
U
0/500