如何访问 Spark Streaming 自定义接收器存储的元数据？答案

【问题标题】：How to access metadata stored by Spark Streaming custom receiver?如何访问 Spark Streaming 自定义接收器存储的元数据？
【发布时间】：2016-02-08 13:50:34
【问题描述】：

Spark Streaming 提供了创建自定义接收器的功能，详细信息 here。要将接收方接收到的数据存储到Spark中，需要使用store(data)方法。

我存储到 Spark 的数据具有与其关联的某些属性。由自定义接收器扩展的Spark Receiver class 提供了几种store(data, metadata) 形式的存储方法，这意味着元数据/属性可以与数据一起存储。下面的代码摘录显示了我如何使用这种方法来存储数据及其元数据/属性。

public class CustomReceiver extends Receiver<String> {

    public CustomReceiver() {
        super(StorageLevel.MEMORY_AND_DISK_2());
    }

    @Override
    public void onStart() {
        new Thread() {
            @Override
            public void run() {
                try {
                    receive();
                } catch (IOException e) {
                    restart("Error connecting: ", e);
                }
            }
        }.start();
    }

    @Override
    public void onStop() {
        // Not needed as receive() method closes resources when stopped
    }

    private void receive() throws IOException {
        String str = getData();
        Map<String, String> metadata = getMetadata();
        Iterator<String> it = Arrays.asList(str.split("\n\r")).iterator();

        store(it, metadata);

        if (isStopped()) {
            closeConnections();
        }
    }
}

这个存储的数据是从另一个类访问的，如以下代码提取所示：

private void testCustomReceiver() {
    JavaDStream<String> custom = ssc.receiverStream(new CustomReceiver());

    JavaDStream<String> processedInput = custom.flatMap(row -> {
        return Arrays.asList(row.split("\\r?\\n"));
    });

    processedInput.print();
}

现在让我们想到我的问题：如何通过上面显示的testCustomReceiver() 方法访问与自定义接收器中的数据一起存储的元数据/属性？

我尝试搜索文档并在调试器中探索JavaDStream 对象以搜索元数据，但无济于事。非常感谢您对此事的任何帮助或建议，谢谢。

【问题讨论】：

你不是已经访问了processedInput.print();的“元数据”
@ab853 您找到解决方案了吗？我也面临同样的问题
@yishaiz，可能不是。这是不久前的事情，但如果我找到了解决方案，我会在这里添加它。

标签： java apache-spark spark-streaming

【解决方案1】：

我一直在研究 Spark 代码，我相信您永远无法再次访问它。事实上，我不相信它曾经被使用过。

Receiver 的主管获取 metadataOption 并将其放入 ReceivedBlockInfo（对 org.apache.spark.streaming 是私有的）。从那里，它去......无处可去。我在流代码库中找不到对 ReceivedBlockInfo.metadataOption 的引用。假设 ReceivedBlockInfo 可能被序列化然后反序列化为不同的类，或者一些时髦的反射检索元数据，但这两种都是反模式，我不会指望它发生。

为什么会在那里？我相信它的意图是让它成为Metadata Checkpointing 系统的一部分，但它要么从未连接，要么接收器元数据和流检查点之间的连接被切断。

无论哪种方式，一旦将块放入流中，块元数据就会消失。

【讨论】：