Java用mapreduce 处理海量行情并转发kafka

Java使用MapReduce处理海量行情数据并转发到Kafka可以采用以下步骤：

1. 编写Mapper类和Reducer类：Mapper类将海量行情数据转换成键值对，其中键为要转发的Kafka主题名称，值为行情数据；Reducer类将相同主题名称的行情数据进行合并，并将结果输出到Kafka。

2. 配置Hadoop集群：配置Hadoop集群，包括HDFS和YARN等组件。将需要处理的海量行情数据上传到HDFS中。

3. 打包Mapper类和Reducer类：将Mapper类和Reducer类打包成jar包。

4. 运行MapReduce任务：使用hadoop命令运行MapReduce任务，将输入文件指定为HDFS中的海量行情数据，将输出文件指定为Kafka主题。

5. 监听Kafka主题：启动一个Kafka消息监听器，监听指定的Kafka主题，并将消息转发给目标消费者。

下面是一个简单的实现示例：

public class QuoteMapper extends Mapper<LongWritable, Text, Text, Text> {

    private Text topic = new Text();
    private Text quote = new Text();

    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        Quote quote = parseQuote(line); // 解析行情数据
        topic.set(quote.getTopic());
        quote.setTimestamp(System.currentTimeMillis());
        this.quote.set(JSON.toJSONString(quote)); // 转换成JSON格式
        context.write(topic, this.quote);
    }
}

public class QuoteReducer extends Reducer<Text, Text, Text, Text> {

    private Producer<String, String> producer;

    public void setup(Context context) throws IOException, InterruptedException {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092"); // Kafka的Broker地址
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        this.producer = new KafkaProducer<>(props);
    }

    public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        for (Text quote : values) {
            ProducerRecord<String, String> record = new ProducerRecord<>(key.toString(), quote.toString());
            this.producer.send(record); // 发送消息到Kafka
        }
    }

    public void cleanup(Context context) throws IOException, InterruptedException {
        this.producer.close();
    }
}

上述代码中，QuoteMapper类将海量行情数据转换成键值对，其中键为要转发的Kafka主题名称，值为行情数据。QuoteReducer类将相同主题名称的行情数据进行合并，并将结果输出到Kafka。在Reducer类的setup()方法中创建Kafka生产者，将消息发送到Kafka中。在Reducer类的cleanup()方法中关闭Kafka生产者。运行MapReduce任务时，使用hadoop命令指定输入文件和输出Kafka主题：

hadoop jar quote-processing.jar QuoteProcessing /input/quotes /output/quotes kafka://localhost:9092

上述命令将HDFS中的/input/quotes文件作为输入，将结果输出到Kafka主题/quotes中。在监听/quotes主题的Kafka消费者中可以获取到转发后的行情数据。