Nebula nGQL查询数据后如何保存为dataframe进行计算？

Hjerry · 2022 年8 月 23 日 10:35

nebula 版本：3.0.0

nebula空间内有很多数据，希望通过nGQL筛选数据后保存为dataframe进行计算（调用nebula Algorithm或者spark计算），不知道应该如何做转df，还请知道的告知一下谢谢

nicole · 2022 年8 月 24 日 01:52

参考spark connector中的数据处理方式，将ngql query 结果Result中的数据进行解析，构造成DataFrame中的Row

vesoft-inc/nebula-spark-connector/blob/6b80a2ff9d481abb89307f8f18ca5580256d5f2d/nebula-spark-connector/src/main/scala/com/vesoft/nebula/connector/reader/NebulaPartitionReader.scala#L109

    
      
              throw new GraphConnectException("storage connect failed.")
            }
            // allocate scanPart to this partition
            val totalPart = metaProvider.getPartitionNumber(nebulaOptions.spaceName)
          
          
  val scanParts = PartitionUtils.getScanParts(index, totalPart, nebulaOptions.partitionNums.toInt)
            LOG.info(s"partition index: ${index}, scanParts: ${scanParts.toString}")
            scanPartIterator = scanParts.iterator
          }
          
          
override def get(): InternalRow = {
            val resultSet: Array[ValueWrapper] =
              dataIterator.next().getValues.toArray.map(v => v.asInstanceOf[ValueWrapper])
            val getters: Array[NebulaValueGetter] = NebulaUtils.makeGetters(schema)
            val mutableRow                        = new SpecificInternalRow(schema.fields.map(x => x.dataType))
          
          
  for (i <- getters.indices) {
              val value: ValueWrapper = resultSet(i)
              var resolved            = false
              if (value.isNull) {
                mutableRow.setNullAt(i)

Hjerry · 2022 年8 月 24 日 10:08

谢谢，那构造dataframe的schema如何获取呢

nicole · 2022 年8 月 24 日 14:31

你通过查询语句查出来的结果里面有每个属性的数据类型的，每个属性是一个ValueWrapper结构，可以通过isLong,isString等方法来判断数据的数据类型，这样根据Nebula中的数据类型将数据转换为SparkSql中的数据类型。

Hjerry · 2022 年8 月 26 日 09:54

spark ReadNebulaConfig方法可以读取nebula数据返回df，但是好像是全量返回的，请问能通过一些方法筛选返回吗

nicole · 2022 年8 月 29 日 02:27

目前扫描数据不能筛选哈，你可以在df上通过sparksql进行筛选

Hjerry · 2022 年8 月 29 日 02:39

全量读转df再筛选感觉效率有点低，目前是通过ngql筛选后直接转df的

nicole · 2022 年8 月 29 日 02:43

也可以的，目前扫描数据的接口上有一个filter参数未实现，实现后你这个需求就很轻松可以满足了

Hjerry · 2022 年8 月 29 日 03:07

加油，快快实现哈哈哈哈

system · 2022 年9 月 28 日 03:07

此话题已在最后回复的 30 天后被自动关闭。不再允许新回复。