java - 如何将reducer的输出写入数据库？

Question

我将从一个例子开始。假设输入数据类似于

User1,product1,time1
User1,product2,time2
User1,product3,time3
User2,product2,time2
User2,product4,time6

现在预期的输出是我必须将数据插入数据库(在我的情况下是 Aerospike(键值存储))，其中数据的格式应为

User1, [ [product1,time1],[product2,time2],[product3,time3] ]
User2, [ [product2,time2],[product4,time6] ]

所以在映射器中我输出以下

UserID, [productid,timestamp]

Please do not assume that [x,y] means i am outputting list i may send data from mappper in any way may be write the data in a custom object

所以在接收方我有格式的数据

User1, [ [product1,time1],[product2,time2],[product3,time3] ]
User2, [ [product2,time2],[product4,time6] ]

现在我可以做两件事

a)我可以编写逻辑以仅在 reducer 中将此数据推送到数据库中
(我不想这样做)

b)我想做的是，当我们执行 Context.write() 时，我希望将数据写入数据库。

请帮助如何做到这一点，并尽可能附上代​​码片段或伪代码

PS:Context.write() 做什么？它写到哪里？它经过了哪些步骤和阶段？

Answer 1

据我了解，调用 context.write 涉及一定数量的步骤

在驱动程序中，我们必须指定输出格式。现在让我们看看如果我们想写入文件会发生什么

为了写入文本文件，我们指定类似

job.setOutputFormatClass(TextOutputFormat.class);

现在，如果我们看到扩展 FileOutputFormat(abstract class) 的 TextOutputFormat 类的实现，它实现了 OutputFormat 接口(interface)，并且 OutputFormat 接口(interface)提供了两个方法

1) getRecordWriter
2) checkOutputSpecs

现在会发生什么，OutputFormatClass 只是告诉您要写入什么样的记录以及记录编写器如何给出，对于记录编写器它只得到 Object Key, Object Value其中 value 可以是单个或列表，并且在记录编写器的实现中，我们指定实际逻辑，例如应该如何编写该记录。

现在回到最初的问题，在我的情况下，应该如何将记录写入数据库 Aerospike

我创建了一个自定义的 OutputFormat 说

public class AerospikeOutputFormat extends OutputFormat {
    //Return a new instance of record writer
    @Override
    public RecordWriter getRecordWriter(TaskAttemptContext context) throws IOException, InterruptedException {
        return new AerospikeRecordWriter(context.getConfiguration(), new Progressable() {
        @Override
        public void progress() {

        }
    });
    }

}

现在我们必须定义一个自定义记录写入器，它将获取一个键和一个值并将数据写入数据库

public class RSRVRecordWriter<KK,VV> extends RecordWriter<KK, VV> {

    @Override
    public void write(KK key, VV value) throws IOException {
        //Now here we can have an instance of aerospikeclient from a singleton class and then we could do client.put()

    }

以上代码只是一个片段，必须采取适当的设计策略。

PS:Aerospike 提供了一个记录器，可以在 this link 上进行扩展以满足您的需求。