使用 Arango-Spark 连接器将 Spark 数据帧保存到 ArangoDB 时出现问题

Question

我正在尝试将 Spark 数据帧保存到 ArangoDB 中，但遇到了一些问题：

1. 保存到 Arango 中的数据（顶点）包含大量关于记录的不需要的“模式”信息

2. 更关键的问题是保存操作会忽略我的“_key”字段并生成随机密钥

这是一个代码sn-p

import com.arangodb.spark.{ArangoSpark, WriteOptions}
import org.apache.spark.sql.SparkSession   
import org.apache.spark.{SparkConf}

object TestGraph extends App{
  println("Hello, test!")

  // set up Spark
  val sparkConf = new SparkConf().setAppName("MySparkDriverApp").setMaster("local[*]").set("spark.executor.memory", "2g")
  val spark = SparkSession.builder().config(sparkConf).getOrCreate()
  val sc = spark.sparkContext

  // set arango options and save vertices
  var options = new WriteOptions()
  options = options.database("test")
  options = options.user("root")
  options = options.password("openSesame")

  val rdd_v = sc.parallelize(Seq(("v1","a"),("v2","b")))
  val df_v = spark.createDataFrame(rdd_v).toDF("_key", "f1")
  df_v.show()

  ArangoSpark.save(df_v, "test_vertex", options)

  println("Test completed")
}

这是数据在 ArangoDB 中的样子（通过 REST API 查询）

{
    "result": [
        {
            "_key": "8607",
            "_id": "test_vertex/8607",
            "_rev": "_Wq0DOVS--_",
            "schema": {
                "_hashCode": 1762032556,
                "bitmap$0": 8,
                "fields": [
                    {
                        "dataType": {
                            "ordering": {
                                "evidence$1$1": {}
                            }
                        },
                        "metadata": {
                            "_hashCode": -1609326920,
                            "bitmap$0": true,
                            "map": {}
                        },
                        "name": "_key",
                        "nullable": true
                    },
                    {
                        "dataType": {
                            "ordering": {
                                "evidence$1$1": {}
                            }
                        },
                        "metadata": {
                            "_hashCode": -1609326920,
                            "bitmap$0": true,
                            "map": {}
                        },
                        "name": "f1",
                        "nullable": true
                    }
                ]
            },
            "values": [
                "v2",
                "b"
            ]
        },
        {
            "_key": "8613",
            "_id": "test_vertex/8613",
            "_rev": "_Wq0DOVW--_",
            "schema": {
                "_hashCode": 1762032556,
                "bitmap$0": 8,
                "fields": [
                    {
                        "dataType": {
                            "ordering": {
                                "evidence$1$1": {}
                            }
                        },
                        "metadata": {
                            "_hashCode": -1609326920,
                            "bitmap$0": true,
                            "map": {}
                        },
                        "name": "_key",
                        "nullable": true
                    },
                    {
                        "dataType": {
                            "ordering": {
                                "evidence$1$1": {}
                            }
                        },
                        "metadata": {
                            "_hashCode": -1609326920,
                            "bitmap$0": true,
                            "map": {}
                        },
                        "name": "f1",
                        "nullable": true
                    }
                ]
            },
            "values": [
                "v1",
                "a"
            ]
        }
    ],
    "hasMore": false,
    "count": 2,
    "cached": false,
    "extra": {
        "stats": {
            "writesExecuted": 0,
            "writesIgnored": 0,
            "scannedFull": 2,
            "scannedIndex": 0,
            "filtered": 0,
            "httpRequests": 0,
            "executionTime": 0.00011944770812988281
        },
        "warnings": []
    },
    "error": false,
    "code": 201
}

有什么建议吗？

Answer 1

保存DataFrame时必须使用方法

ArangoSpark.saveDF(DataFrame, String, WriteOptions)

而不是

ArangoSpark.save(Dataset, String, WriteOptions)