0

正如 spark docs 所说，它支持 kafka 作为数据流源。但是我使用 ZeroMQ，并且没有 ZeroMQUtils。所以我该如何使用它？一般来说，其他MQ怎么样。我对火花和火花流完全陌生，所以如果问题很愚蠢，我很抱歉。谁能给我一个解决方案。谢谢顺便说一句，我使用 python。

更新，我终于用自定义接收器在 java 中做到了。以下是我的解决方案

public class ZeroMQReceiver extends Receiver<T> {

    private static final ObjectMapper mapper = new ObjectMapper();

    public ZeroMQReceiver() {

        super(StorageLevel.MEMORY_AND_DISK_2());
    }

    @Override
    public void onStart() {
        // Start the thread that receives data over a connection
        new Thread(this::receive).start();
    }

    @Override
    public void onStop() {
        // There is nothing much to do as the thread calling receive()
        // is designed to stop by itself if isStopped() returns false
    }

    /** Create a socket connection and receive data until receiver is stopped */
    private void receive() {
        String message = null;

        try {

            ZMQ.Context context = ZMQ.context(1); 
            ZMQ.Socket subscriber = context.socket(ZMQ.SUB);     
            subscriber.connect("tcp://ip:port");    
            subscriber.subscribe("".getBytes());  

            // Until stopped or connection broken continue reading
            while (!isStopped() && (message = subscriber.recvStr()) != null) {
                List<T> results = mapper.readValue(message,
                        new TypeReference<List<T>>(){} );
                for (T item : results) {
                    store(item);
                }
            }
            // Restart in an attempt to connect again when server is active again
            restart("Trying to connect again");
        } catch(Throwable t) {
            // restart if there is any other error
            restart("Error receiving data", t);
        }
    }
}
4

1 回答 1

0

我假设您在谈论结构化流。

我不熟悉 ZeroMQ，但 Spark Structured Streaming 源中的一个重点是可重播性（以确保容错），如果我理解正确，ZeroMQ 不会提供开箱即用的功能。

一种实用的方法是在 Kafka 中缓冲数据并使用 KafkaSource 或作为（本地 FS/NFS、HDFS、S3）目录中的文件并使用 FileSource 进行读取。参照。星火文档。如果您使用 FileSource，请确保不要将任何内容附加到 FileSource 的输入目录中的现有文件，而是自动将它们移动到目录中。

于 2018-11-09T13:08:15.230 回答