Spark Dataframe：选择不同的行

Question

我尝试了两种方法从镶木地板中找到不同的行，但它似乎不起作用。

尝试1： Dataset<Row> df = sqlContext.read().parquet("location.parquet").distinct();

但是抛出

Cannot have map type columns in DataFrame which calls set operations
(intersect, except, etc.),
but the type of column canvasHashes is map<string,string>;;

尝试 2： 尝试运行 sql 查询：

 Dataset<Row> df = sqlContext.read().parquet("location.parquet");
    rawLandingDS.createOrReplaceTempView("df");
    Dataset<Row> landingDF = sqlContext.sql("SELECT distinct on timestamp * from df");

我得到的错误：

 = SQL ==
SELECT distinct on timestamp * from df
-----------------------------^^^

有没有办法在读取镶木地板文件时获得不同的记录？我可以使用的任何读取选项。

原文由 Himanshu Yadav 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 1

您面临的问题在异常消息中明确说明 - 因为 MapType 列既不可散列也不可排序，不能用作分组或分区表达式的一部分。

您对 SQL 解决方案的看法在逻辑上不等同于 distinct 上的 Dataset 。如果您想基于一组兼容列删除重复数据，您应该使用 dropDuplicates ：

 df.dropDuplicates("timestamp")

这相当于

SELECT timestamp, first(c1) AS c1, first(c2) AS c2,  ..., first(cn) AS cn,
       first(canvasHashes) AS canvasHashes
FROM df GROUP BY timestamp

不幸的是，如果您的目标是实际的 DISTINCT 这不会那么容易。可能的解决方案是利用 Scala* Map 散列。您可以像这样定义 Scala udf ：

 spark.udf.register("scalaHash", (x: Map[String, String]) => x.##)

然后在您的 Java 代码中使用它来派生可用于 dropDuplicates 的列：

  df
  .selectExpr("*", "scalaHash(canvasHashes) AS hash_of_canvas_hashes")
  .dropDuplicates(
    // All columns excluding canvasHashes / hash_of_canvas_hashes
    "timestamp",  "c1", "c2", ..., "cn"
    // Hash used as surrogate of canvasHashes
    "hash_of_canvas_hashes"
  )

与 SQL 等效

SELECT
  timestamp, c1, c2, ..., cn,   -- All columns excluding canvasHashes
  first(canvasHashes) AS canvasHashes
FROM df GROUP BY
  timestamp, c1, c2, ..., cn    -- All columns excluding canvasHashes

---

\* 请注意 java.util.Map 和它的 hashCode 将不起作用，因为 hashCode 不一致。

原文由 user10938362 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 2

是的，语法不正确，应该是：

 Dataset<Row> landingDF = sqlContext.sql("SELECT distinct * from df");

原文由 Andronicus 发布，翻译遵循 CC BY-SA 4.0 许可协议