mysql - 通过 Kafka 和 Spark 消费大数据

Question

我在 Json 中有一个由 Websocket 提供的流数据，其大小在每秒 1MB 到 60MB 之间变化。

我必须解码数据然后解析它，最后写入 mysql。

我想到了 2 个想法:

1)从Socket中读取数据，解码后在Producer中通过Avro发送给Consumer，然后在Spark map上获取数据写入mysql，在Consumer中reduce

2)从Socket中读取数据，然后将数据发送给Producer中的Consumer，然后在Consumer中获取数据，然后在Spark上解码，将解析后的数据发送给Spark Job写入mysql。

你有什么想法吗？

制作人

/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package com.tan;


import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.util.Properties;


import java.util.stream.Stream;

import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
/**
 *
 * @author Tan
 */
public class MainKafkaProducer {

    /**
     * @param args the command line arguments
     */
    public static void main(String[] args) throws InterruptedException {
        // TODO code application logic here
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        //props.put("group.id", "mygroup");
        //props.put("max.partition.fetch.bytes", "100000000");
        //props.put("serializer.class", "kafka.serializer.StringEncoder");
        //props.put("partitioner.class","kafka.producer.DefaultPartitioner");
        //props.put("request.required.acks", "1");

         KafkaProducer<String, String> producer = new KafkaProducer<>(props);

         // Read the data from websocket and send it to consumer
         //for (int i = 0; i < 100; i++) {
            String fileName = "/Users/Tan/Desktop/feed.json";
            try{
                BufferedReader file = new BufferedReader(new FileReader(fileName));
                String st = file.readLine();
                for(int i = 0; i < 100; i++)
                {
                    ProducerRecord<String, String> record = new ProducerRecord<>("mytopic", st);
                    producer.send(record);
                }
            }catch(IOException e){
                e.printStackTrace();
            }
        //}

        /*
        for(int i = 0; i < 100; i++)
        {
            ProducerRecord<String, String> record2 = new ProducerRecord<>("mytopic", "Hasan-" + i);
            producer.send(record2);
        }
        */


        producer.close();
    }

}

消费者

/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package com.tan;

import kafka.serializer.DefaultDecoder;
import kafka.serializer.StringDecoder;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaPairInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka.KafkaUtils;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
/**
 *
 * @author Tan
 */
public class MainKafkaConsumer {
    /**
     * @param args the command line arguments
     */
    public static void main(String[] args) {

        SparkConf conf = new SparkConf()
                .setAppName(MainKafkaConsumer.class.getName())
                .setMaster("local[*]");
        JavaSparkContext sc = new JavaSparkContext(conf);
        JavaStreamingContext ssc = new JavaStreamingContext(sc, new Duration(2000));

        Set<String> topics = Collections.singleton("mytopic");
        Map<String, String> kafkaParams = new HashMap<>();
        kafkaParams.put("metadata.broker.list", "localhost:9092");

        JavaPairInputDStream<String, String> directKafkaStream = KafkaUtils.createDirectStream(ssc, 
                String.class, String.class, 
                StringDecoder.class, StringDecoder.class, 
                kafkaParams, topics);

        directKafkaStream.foreachRDD(rdd -> {

            rdd.foreach(records -> {

                System.out.println(records._2);

            });

        });
        /*
        directKafkaStream.foreachRDD(rdd -> {
            System.out.println("--- New RDD with " + rdd.partitions().size()
                    + " partitions and " + rdd.count() + " records");
            rdd.foreach(record -> {
                System.out.println(record._2);
            });
        });
        */



        ssc.start();
        ssc.awaitTermination();

    }

}

Answer 1

你的过程很好，重点是avro转换。您的数据不是那么大，1Mb 到 60Mb。

这里我有一个类似的过程，从 MQ 读取数据，处理数据，转换为 avro，发送到 kafka，从 kafka 消费，解析数据并在其他 MQ 中发布。

当我们的数据很大时，比如 >= 1Gb，Avro 会帮上大忙。但在某些情况下，我们的数据非常小，如 < 10Mb。在这种情况下，Avro 使我们的处理速度有点慢，网络传输没有任何好处。

我对你的建议是，如果你的网络足够好，不能转换成 avro，最好不要转换成 avro。为了提高 Spark Side 的性能，请使用大量分区配置 kafka 主题，因为如果只有一个分区，则 spark 将无法正确进行并行化。检查this对您有帮助的文字。