Spark Couchbase 连接器 - N1QL RDD 到 DataFrame

Question

我正在尝试将 RDD 形式的 couchbase 转换为 DataFrame（scala 2.11 - 和 spark 2.1），但出现重载错误，我的代码如下，有什么想法吗？另一个帖子没有完全回答这个问题。

我在 Databricks 笔记本中执行此操作，并且我将沙发连接器用于纯 DataFrame，但如果我想做一个客户 N1QL 查询，一些更定制的，这是我能想到的最好的，首先使用 RDD？

首先，有没有更好的方法在本机 Dataframe 中执行此查询？我想我需要使用 n1qL 和 RDD 还是我在这里遗漏了什么？

请让我知道我在下面的 RDD 转换代码中做错了什么，我还得到了 :84: error: 重载方法值 createDataFrame with alternatives: error....谢谢！

val reconciliationSchema = 
   new StructType()
      .add("numEvents", IntegerType)
      .add("eventCategory", StringType)
      .add("eventName", StringType)

val orderEventsCouchbaseQuery = """
  SELECT 
    count(*) as numEvents, event.eventCategory, event.eventName
  FROM 
    events
  WHERE 
    STR_TO_UTC(event.eventOccurredTime)
      BETWEEN STR_TO_UTC("2017-06-16") AND STR_TO_UTC("2017-06-26")
  GROUP BY event.eventCategory, event.eventName
  order by event.eventCategory, event.eventName
"""

val queryResultRDD = sc.couchbaseQuery(N1qlQuery.simple(orderEventsCouchbaseQuery),"events").map(_.value)
val queryResultDF: DataFrame = spark.createDataFrame(queryResultRDD,reconciliationSchema)
display(queryResultDF)

Answer 1

我认为您遇到的问题与其说是与沙发底座相关的问题，不如说是 spark/scala 类型的推理问题。当您使用createDataFrame 时，在这种情况下，spark 需要使用Row，而不是使用该 rdd 的 couchbase 查询的返回类型。

所以这里有一些类似的示例代码，你可以看到当变成一行时它工作正常：

val query = N1qlQuery.simple("" +
      "select country, count(*) as count " +
      "from `travel-sample` " +
      "where type = 'airport' " +
      "group by country " +
      "order by count desc")

val schema = StructType(
        StructField("count", IntegerType) ::
        StructField("country", StringType) :: Nil
    )

val rdd = spark.sparkContext.couchbaseQuery(query).map(r => Row(r.value.getInt("count"), r.value.getString("country")))
spark.createDataFrame(rdd, schema).show()