使用 Java Client Api(DMSDK) 从 marklogic 创建 CSV 文件

Question

我想从我的marklogic db为1.3M记录创建一个csv文件。我尝试为此使用 CORB，但花费的时间比我预期的要长。 我的数据是这样的

{
"One": {
"Name": "One",
"Country": "US"
}, 
"Two": {
"State": "kentucky"
}, 
"Three": {
"Element1": "value1", 
"Element2": "value2", 
"Element3": "value3", 
"Element4": "value4",
so on ...
}
}

以下是我的 Corb 模块

选择器.xqy

var total = cts.uris("", null, cts.collectionQuery("data"));
fn.insertBefore(total,0,fn.count(total))

Transform.xqy(我将所有元素保存在数组中)

var name = fn.tokenize(URI, ";");
const node = cts.doc(name);
var a= node.xpath("/One/*");
var b= node.xpath("/Two/*");
var c= node.xpath("/Three/*");
fn.stringJoin([a, b, c,name], " , ")

我的属性文件

THREAD-COUNT=16
BATCH-SIZE=1000
URIS-MODULE=selector.sjs|ADHOC
PROCESS-MODULE=transform.sjs|ADHOC
PROCESS-TASK=com.marklogic.developer.corb.ExportBatchToFileTask
EXPORT-FILE-NAME=Report.csv
PRE-BATCH-TASK=com.marklogic.developer.corb.PreBatchUpdateFileTask
EXPORT-FILE-TOP-CONTENT=Col1,col2,....col16 -- i have 16 columns 

创建 csv 文件花了 1 个多小时。此外，为了在集群中尝试，我需要先配置负载均衡器。而 Java 客户端 api 将在没有任何负载均衡器的情况下在所有节点之间分配工作。

如何在Java客户端APi中实现相同的功能，我知道可以使用ServerTransform和ApplyTransformListener触发转换模块。

public static void main(String[] args) {
// TODO Auto-generated method stub
DatabaseClient client = DatabaseClientFactory.newClient
("localhost", pwd, "x", "x",  DatabaseClientFactory.Authentication.DIGEST);
ServerTransform txform = new ServerTransform("tsm"); -- Here i am implementing same logic of above `tranform module` .
QueryManager qm = client.newQueryManager();
StructuredQueryBuilder query = qm.newStructuredQueryBuilder();
query.collection();
DataMovementManager dmm = client.newDataMovementManager();
QueryBatcher batcher = dmm.newQueryBatcher(query.collections("data"));
batcher.withBatchSize(2000)
.withThreadCount(16)
.withConsistentSnapshot()
.onUrisReady(
new ApplyTransformListener().withTransform(txform))
.onBatchSuccess(batch-> {
System.out.println(
batch.getTimestamp().getTime() +
" documents written: " +
batch.getJobWritesSoFar());
})
.onBatchFailure((batch,throwable) -> {
throwable.printStackTrace();
});
// start the job and feed input to the batcher
dmm.startJob(batcher);
batcher.awaitCompletion();
dmm.stopJob(batcher);
client.release();
}

但是我怎样才能像在 CORB 中那样发送 csv 文件头(即导出文件顶部内容)。是否有任何用于实现CSV文件的文档？哪个类将实现它？

任何帮助不胜感激

谢谢

Answer 1

可能最简单的选择是ml-gradle将数据导出到CSV，它在后台使用Java客户端API和DMSDK。

请注意，您可能希望安装服务器端 REST 转换以仅提取 CSV 输出中所需的数据，而不是下载整个文档内容然后在 Java 端提取。

有关使用 DMSDK 和创建聚合 CSV(一个 CSV 用于所有记录)所需的代码的工作示例，请参阅 ExporToWriterListenerTest.testMassExportToWriter。 为了 SO，以下是关键代码片段(有一些小的简化更改，包括编写列标题(未经测试的代码))：

try (FileWriter writer = new FileWriter(outputFile)) {
writer.write("uri,collection,contents");
writer.flush();
ExportToWriterListener exportListener = new ExportToWriterListener(writer)
.withRecordSuffix("n")
.withMetadataCategory(DocumentManager.Metadata.COLLECTIONS)
.onGenerateOutput(
record -> {
String uri = record.getUri();
String collection = record.getMetadata(new DocumentMetadataHandle()).getCollections().iterator().next();
String contents = record.getContentAs(String.class);
return uri + "," + collection + "," + contents;
}
);
QueryBatcher queryJob =
moveMgr.newQueryBatcher(query)
.withThreadCount(5)
.withBatchSize(10)
.onUrisReady(exportListener)
.onQueryFailure( throwable -> throwable.printStackTrace() );
moveMgr.startJob( queryJob );
queryJob.awaitCompletion();
moveMgr.stopJob( queryJob );
}

但是，除非您知道您的内容没有双引号、换行符或非 ASCII 字符，否则建议使用 CSV 库以确保您的输出正确转义。 要使用CSV库，您当然可以使用库的任何教程。 无需担心线程安全性，因为 ExportToWriterListener 在同步块中运行侦听器，以防止对编写器的重叠写入。 下面是使用一个CSV库Jackson CsvMapper的示例。

请注意，您不必使用 ExportToWriterListener . . .您可以将其用作编写自己的侦听器的起点。 特别是，由于您主要关注的是性能，因此您可能希望让侦听器为每个线程写入一个文件，然后进行后处理以将内容组合在一起。 这取决于你。