不支持 SparkSQL 表创建 CLUSTER BY答案

【问题标题】：SparkSQL table creation CLUSTER BY not supported不支持 SparkSQL 表创建 CLUSTER BY
【发布时间】：2017-06-15 23:09:50
【问题描述】：

根据 Spark 文档 https://spark.apache.org/docs/2.1.0/sql-programming-guide.html#supported-hive-features，支持 hive 语句 CLUSTER BY。但是当我尝试使用以下直线查询创建表时

CREATE TABLE set_bucketing_test (key INT, value STRING) CLUSTERED BY (key) INTO 10 BUCKETS;

我收到以下错误

Error: org.apache.spark.sql.catalyst.parser.ParseException:
Operation not allowed: CREATE TABLE ... CLUSTERED BY(line 1, pos 0)

不知道我在做什么错误。有什么帮助吗？

【问题讨论】：

Hive bucketing is not supported by spark。这是预期的行为。
那么创建CLUSTERED表的查询应该是什么？

标签： hive apache-spark-sql hiveql

【解决方案1】：

您可以利用 spark-sql 中的集群功能进行表创建、表连接等，它充当配置单元以避免数据交换和 spark2.1+ 中的排序

见https://issues.apache.org/jira/browse/SPARK-15453

目前 hive 无法识别此功能，因为元数据在 spark 和 hive 之间不兼容，这就是为什么即使在 hive 端识别此表也不能使用相同的语法，这会将所有列视为array

以下示例可能会给您一些想法：

准备源码

val df = (0 until 80000).map(i => (i, i.toString, i.toString)).toDF("item_id", "country", "state").coalesce(1)

从源创建两个桶表

您会看到“与 Hive 不兼容”。向右滚动

df.write.bucketBy(100, "country", "state").sortBy("country", "state").saveAsTable("kofeng.lstg_bucket_test")

17/03/13 15:12:01 WARN HiveExternalCatalog: Persisting bucketed data source table `kofeng`.`lstg_bucket_test` into Hive metastore in Spark SQL specific format, which is NOT compatible with Hive.

df.write.bucketBy(100, "country", "state").sortBy("country", "state").saveAsTable("kofeng.lstg_bucket_test2")

加入他们并解释

由于音量小，请先禁用广播加入。

import org.apache.spark.sql.SparkSession
val spark = SparkSession.builder().appName("Spark SQL basic example").config("spark.sql.autoBroadcastJoinThreshold", "0").getOrCreate()

计划在SPARK 2.1.0中避免交换和排序，在SPARK2.0中避免交换，只过滤和扫描证明数据局部性利用。

 val query = """
 |SELECT *
 |FROM
 |  kofeng.lstg_bucket_test a
 |JOIN
 |  kofeng.lstg_bucket_test2 b
 |ON a.country=b.country AND
 |   a.state=b.state
      """.stripMargin
val joinDF = sql(query)


    scala> joinDF.queryExecution.executedPlan
    res10: org.apache.spark.sql.execution.SparkPlan =
*SortMergeJoin [country#71, state#72], [country#74, state#75], Inner
:- *Project [item_id#70, country#71, state#72]
:  +- *Filter (isnotnull(country#71) && isnotnull(state#72))
:     +- *FileScan parquet kofeng.lstg_bucket_test[item_id#70,country#71,state#72] Batched: true, Format: Parquet, Location: InMemoryFileIndex[hdfs://ares-lvs-nn-ha/user/hive/warehouse/kofeng.db/lstg_bucket_test], PartitionFilters: [], PushedFilters: [IsNotNull(country), IsNotNull(state)], ReadSchema: struct<item_id:int,country:int,state:string>
+- *Project [item_id#73, country#74, state#75]
   +- *Filter (isnotnull(country#74) && isnotnull(state#75))
      +- *FileScan parquet kofeng.lstg_bucket_test2[item_id#73,country#74,state#75] Batched: true, Format: Parquet...

【讨论】：