如何将 JavaRDD 写入 marklogic 数据库

Question

我正在使用 marklogic 数据库评估 spark。我已经阅读了一个 csv 文件，现在我有一个 JavaRDD 对象，我必须将其转储到 marklogic 数据库中。

    SparkConf conf = new SparkConf().setAppName("org.sparkexample.Dataload").setMaster("local");
    JavaSparkContext sc = new JavaSparkContext(conf);

    JavaRDD<String> data = sc.textFile("/root/ml/workArea/data.csv");
    SQLContext sqlContext = new SQLContext(sc);
    JavaRDD<Record> rdd_records = data.map(
      new Function<String, Record>() {
          public Record call(String line) throws Exception {
             String[] fields = line.split(",");
             Record sd = new Record(fields[0], fields[1], fields[2], fields[3],fields[4]);
             return sd;
          }
    });

我想将这个 JavaRDD 对象写入 marklogic 数据库。

是否有任何 spark api 可用于更快地写入 marklogic 数据库？

比方说，如果我们不能将 JavaRDD 直接写入 marklogic，那么实现这一点的正确方法是什么？

这是我用来将 JavaRDD 数据写入 marklogic 数据库的代码，如果这样做是错误的方法，请告诉我。

final DatabaseClient client = DatabaseClientFactory.newClient("localhost",8070, "MLTest");
    final XMLDocumentManager docMgr = client.newXMLDocumentManager();   
    rdd_records.foreachPartition(new VoidFunction<Iterator<Record>>() {
        public void call(Iterator<Record> partitionOfRecords) {
            while (partitionOfRecords.hasNext()) {
                Record record = partitionOfRecords.next();
                System.out.println("partitionOfRecords - "+record.toString());
                String docId = "/example/"+record.getID()+".xml";
                JAXBContext context = JAXBContext.newInstance(Record.class);
                JAXBHandle<Record> handle = new JAXBHandle<Record>(context);
                handle.set(record);
                docMgr.writeAs(docId, handle);
            }
      }
    });
    client.release();

我已经使用 java 客户端 api 来编写数据，但是即使 POJO 类 Record 正在实现 Serializable 接口，我也遇到了异常。请让我知道可能是什么原因以及如何解决。

org.apache.spark.sparkexception 任务不可序列化。

Answer 1

将数据输入 MarkLogic 的最简单方法是通过 HTTP 和客户端 REST API - 特别是 /v1/documents 端点 - http://docs.marklogic.com/REST/client/management。

有多种方法可以优化这一点，例如通过写入集，但根据您的问题，我认为首先要决定的是 - 您想为每条记录编写什么样的文档？您的示例在 CSV 中显示了 5 列 - 通常，您将编写包含 5 个字段/元素的 JSON 或 XML 文档，每个字段/元素均基于列索引命名。因此，您需要编写一些代码来生成该 JSON/XML，然后使用您喜欢的任何 HTTP 客户端（一种选择是 MarkLogic Java 客户端 API）将该文档写入 MarkLogic。

这解决了您如何将 JavaRDD 写入 MarkLogic 的问题 - 但如果您的目标是尽快将数据从 CSV 获取到 MarkLogic，那么请跳过 Spark 并使用 mlcp - https://docs.marklogic.com/guide/mlcp/import#id_70366 - 这涉及零编码。

Answer 2

修改了spark streaming guide的例子，在这里你必须实现连接和编写特定于数据库的逻辑。

public void send(JavaRDD<String> rdd) {
    rdd.foreachPartition(new VoidFunction<Iterator<String>>() {
      @Override
      public void call(Iterator<String> partitionOfRecords) {
        // ConnectionPool is a static, lazily initialized pool of
        Connection connection = ConnectionPool.getConnection();
        while (partitionOfRecords.hasNext()) {
          connection.send(partitionOfRecords.next());
        }
        ConnectionPool.returnConnection(connection); // return to the pool
        // for future reuse
      }
    });
  }

Answer 3

我想知道您是否只需要确保您在 VoidFunction 内部访问且在其外部实例化的所有内容都是可序列化的（请参阅this page）。 DatabaseClient 和 XMLDocumentManager 当然不可序列化，因为它们是连接的资源。但是，不要在 VoidFunction 中实例化 DatabaseClient 是对的，因为那样效率会降低（尽管它会起作用）。我不知道以下想法是否适用于火花。但我猜你可以创建一个持有单例 DatabaseClient 实例的类：

public static class MLClient {
  private static DatabaseClient singleton;
  private MLClient() {}

  public static DatabaseClient get(DatabaseClientFactory.Bean connectionInfo) {
    if ( connectionInfo == null ) {
      throw new IllegalArgumentException("connectionInfo cannot be null");
    }
    if ( singleton == null ) {
      singleton = connectionInfo.newClient();
    }
    return singleton;
  }
}

然后您只需在您的 VoidFunction 之外创建一个可序列化的 DatabaseClientFactory.Bean，这样您的身份验证信息仍然是集中的

DatabaseClientFactory.Bean connectionInfo = 
  new DatabaseClientFactory.Bean();
connectionInfo.setHost("localhost");
connectionInfo.setPort(8000);
connectionInfo.setUser("admin");
connectionInfo.setPassword("admin");
connectionInfo.setAuthenticationValue("digest");

然后在您的 VoidFunction 中，您可以像这样获得单例 DatabaseClient 和新的 XMLDocumentManager：

DatabaseClient client = MLClient.get(connectionInfo);
XMLDocumentManager docMgr = client.newXMLDocumentManager();