java - 如何在 Java 中迭代 Spark 数据集并更新列值？

Question

我正在研究 POC，我必须使用 token 更新数据库中的帐号。我将数据读入数据集dsRecords(大约2M条记录)。我有另一个例程，它捕获了不同的帐号并获取了 token ，映射存储在 HashMap 中。

Dataset<Row> applySwappedTokens(Dataset<Row> dsRecords, Map<String, String> mappedTokens){
}

现在，我必须迭代数据集才能执行以下操作 -1.读取帐号列(accountNumber)值并使用mappedTokens中的 token 值更新(我知道数据集是不可变的。因此，更新数据集意味着创建具有更新行的数据集副本)。这可以通过 JOIN 或其他操作来实现，但由于第二项任务，我没有在这方面花费精力。2. 读取另一个 XML blob 列并找到帐号并更新它。

到目前为止，我尝试过的所有选项都会因不可序列化的代码而导致编译时错误或测试编译错误。大多数在线资源都是 Scala 语言而不是 Java 语言。请帮忙。

Spark 2.1Java 8

方法1 - 由于序列化错误而无法测试。

Dataset<Row> output = sparkSession.sqlContext().createDataFrame(dsRecords.javaRDD().map(row ->  {
            return RowFactory.create(row.get(0), row.get(1), row.get(2), swapToken(row.get(3)),row.get(4));
        }), dsRecords.schema());

        return output;

String swapToken(Object inputToken) {
        return mappedTokens.get(inputToken);//mappedToken will have to be instance field.
    }

方法2-不完整。

dsRecords.foreach((ForeachFunction<Row>) row -> {
            Integer index = row.fieldIndex("accountNumber");
            String pan = row.getString(index);
            String swap = this.swapToken(pan);
            //TODO: create a dataset with rows from dsRecords but swap value.

        });

方法3 - 使用带有 map 功能的UDF

创建一个 UDF2(采用 2 个输入参数，即 accountNumber 和mappedToken 并返回 token )。看来UDF只能取列值

更新 1 - UDF所以，我实现了UDF(AFK，稍后会发布代码):1.定义UDF1‘updateToken’来传递xml列值并返回更新后的xml值。2. 将具有帐户- token 对映射的 HashMap 实例“mappedTokens”设为静态。在我的 UDF 函数内访问它，以在 xml 字符串中查找帐户并使用 token 进行更新。

我可以测试我的 applySwappedTokens 函数，该函数在数据集“withColumn”上调用上述 UDF。然而，当我运行 Spark 程序时，我看到“mappedToken”具有“null”数据，因此 xml 列会使用空数据进行更新。我认为静态“mappedTokens”要么在另一个 JVM 中，要么在驱动程序中(即使在本地，spark 也会创建隔离的驱动程序、执行程序)。令人沮丧的是，没有简单的解决方案来迭代和更新 Spark 中的行。

Dataset<Row> processByRow(Dataset<Row> dsRecords, SparkSession sparkSession) {
        sparkSession.udf().register("updateToken", updateToken, DataTypes.StringType);          
        return ds = dsRecords.withColumn("eventRecordTokenText", callUDF("updateToken", dsRecords.col("eventRecordTokenText")));
            }

static UDF1 updateToken = new UDF1<String, String>() {
        public String call(final String tokenText) throws Exception {
                // xml operations here..
                for (int nodeIndex = 0; nodeIndex < nList.getLength(); nodeIndex++) {
                    Node thisNode = nList.item(nodeIndex);
                    if (thisNode.getAttributes().getNamedItem("ProcessTokenValue") != null && thisNode.getAttributes()
                            .getNamedItem("ProcessTokenValue").getNodeValue().equalsIgnoreCase("true")) {
                        Node valueNode = thisNode.getAttributes().getNamedItem("Value");
                        String thisToken = valueNode.getNodeValue();
                        String newToken = mappedTokens.get(thisToken); // *returns null values from the map*
                        if(newToken != null && !newToken.isEmpty())
                        valueNode.setNodeValue(newToken);
                    }
                }
                // more xml operations here..
                return output;
        }
    };

更新 2 - 迭代和更新现在，我正在尝试逐行遍历..

Dataset<Row> processByRow1(Dataset<Row> dsRecords, SparkSession sparkSession) {
        List<MongoRecordSmall> newRows = new ArrayList<MongoRecordSmall>();
            dsRecords.foreach((ForeachFunction<Row>) record -> {
            String currentToken = record.getAs(AppConstants.TokenCol);
            String newToken = mappedTokens.get(currentToken);
            newRows.add(new MongoRecordSmall(record.getString(0), record.getString(1), newToken, record.getString(3)));
            logger.error(“Size plus=“+newRows.size());
        });
       return sparkSession.createDataFrame(newRows, MongoRecordSmall.class);
    }

这会引发序列化错误。似乎( https://databricks.gitbooks.io/databricks-spark-knowledge-base/content/troubleshooting/javaionotserializableexception.html )我的类存在上述逻辑，正在被序列化并发送到工作节点，但未能这样做。

Answer 1

由于我没有找到更好的答案，我将用我实现的解决方案来回答我的问题(它看起来很难看!)-

Dataset<Row> processByRowUpdate(Dataset<Row> dsRecords, SparkSession sparkSession) {
        List<Row> rows = dsRecords.select("accountIdentifier", "accountNumber").collectAsList();
List<MongoRecord> newRows = new ArrayList<MongoRecord>();
        ListIterator<Row> it = rows.listIterator();
        boolean errorOccurred = false;
        while (it.hasNext()) {
            try {
                Row record = it.next();
                MongoRecord mongo = new MongoRecord();
       if (!record.isNullAt(record.fieldIndex("accountIdentifier")))
                    mongo.setAccountIdentifier(String.valueOf(record.getDouble(record.fieldIndex("accountIdentifier"))));

           //... and so on
       newRows.add(mongo);
       } catch (Exception exception) {}
sparkSession.createDataFrame(newRows, MongoRecord.class);